From 08fe26a0bc1b7a11d74d1cf778cb64508438d7c4 Mon Sep 17 00:00:00 2001 From: Saish Shinde Date: Fri, 10 Jul 2026 13:19:26 +0530 Subject: [PATCH] release: harden v0.9.2 product qualification --- .github/workflows/ci.yml | 5 + .github/workflows/product-qualification.yml | 44 +++ .github/workflows/release.yml | 90 ++++- .gitignore | 1 + CHANGELOG.md | 11 + README.md | 13 + docs/PUBLISH.md | 126 +++---- docs/QUALITY_GATES.md | 56 +++ pyproject.toml | 35 +- scm/__init__.py | 12 +- scripts/installed_wheel_smoke.py | 154 ++++++++ scripts/run_product_qualification.py | 275 ++++++++++++++ sdk/js/package.json | 2 +- sdk/js/test/live-smoke.mjs | 36 ++ src/api/demo_router.py | 1 + src/api/main.py | 34 +- src/cli/main.py | 28 +- src/cloud/accounts.py | 46 ++- src/core/sqlite_db.py | 17 +- src/integrations/mcp_server.py | 347 +++++++++++++----- src/integrations/memories_api.py | 127 +++++-- src/integrations/tools.py | 92 +++-- src/integrations/user_state_store.py | 200 ++++++++++ src/sleep/forgetting_dynamics.py | 150 +++++++- src/sleep/sleep_cycle.py | 34 +- src/version.py | 3 + .../agent_with_tools/test_supervisor_team.py | 6 + tests/production/__init__.py | 1 + tests/production/conftest.py | 50 +++ tests/production/test_api_abuse.py | 104 ++++++ tests/production/test_concurrency.py | 152 ++++++++ tests/production/test_contract.py | 119 ++++++ tests/production/test_performance.py | 97 +++++ tests/production/test_recovery.py | 80 ++++ tests/production/test_security.py | 111 ++++++ tests/test_ab_hierarchical.py | 10 +- tests/test_crazy_brutal.py | 5 +- tests/test_wake_summary_e2e.py | 16 +- 38 files changed, 2388 insertions(+), 302 deletions(-) create mode 100644 .github/workflows/product-qualification.yml create mode 100644 docs/QUALITY_GATES.md create mode 100644 scripts/installed_wheel_smoke.py create mode 100644 scripts/run_product_qualification.py create mode 100644 sdk/js/test/live-smoke.mjs create mode 100644 src/integrations/user_state_store.py create mode 100644 src/version.py create mode 100644 tests/production/__init__.py create mode 100644 tests/production/conftest.py create mode 100644 tests/production/test_api_abuse.py create mode 100644 tests/production/test_concurrency.py create mode 100644 tests/production/test_contract.py create mode 100644 tests/production/test_performance.py create mode 100644 tests/production/test_recovery.py create mode 100644 tests/production/test_security.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8235b74..39a29ec 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,11 +31,14 @@ jobs: - name: Run focused regression suite env: + SCM_DATA_DIR: /tmp/scm-ci-data-${{ matrix.python-version }} LLM_PROVIDER: "" SCM_EMBEDDING_BACKEND: hash SCM_AUTO_SLEEP_DISABLE: "1" + IDLE_LEARNER_ENABLED: "false" run: | pytest \ + tests/production \ tests/test_scm_sdk.py \ tests/test_product_runtime_api.py \ tests/test_mcp_contract.py \ @@ -148,6 +151,7 @@ jobs: import importlib.resources as resources from scm import SCMClient, SCMEngine + assert SCMClient.__name__ == "SCMClient" assert resources.files("src.core").joinpath("locales/en.json").is_file() assert resources.files("src.api").joinpath("static/app.html").is_file() @@ -248,3 +252,4 @@ jobs: cd sdk/js npm test npm pack --dry-run + npm pack --dry-run diff --git a/.github/workflows/product-qualification.yml b/.github/workflows/product-qualification.yml new file mode 100644 index 0000000..d9eeb47 --- /dev/null +++ b/.github/workflows/product-qualification.yml @@ -0,0 +1,44 @@ +name: Product Qualification + +on: + push: + branches: [main] + pull_request: + branches: [main] + workflow_dispatch: + +jobs: + release-gate: + name: Adversarial release gate + runs-on: ubuntu-latest + timeout-minutes: 30 + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - uses: actions/setup-node@v4 + with: + node-version: "20" + + - name: Install qualification dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" build twine + + - name: Run credential-free product qualification + run: | + python scripts/run_product_qualification.py \ + --output product-qualification.json + + - name: Upload qualification evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: product-qualification + path: product-qualification.json + if-no-files-found: error diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 7532d66..0b4cddf 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -1,16 +1,73 @@ -name: Publish to PyPI +name: Release to PyPI on: + push: + tags: + - "v*.*.*" workflow_dispatch: inputs: version: - description: "Version tag (must match pyproject.toml)" + description: "Existing version tag to publish, for example v0.9.2" required: true + type: string jobs: + qualify: + name: Product release qualification + runs-on: ubuntu-latest + timeout-minutes: 30 + + steps: + - uses: actions/checkout@v4 + with: + ref: ${{ github.event_name == 'workflow_dispatch' && inputs.version || github.ref }} + + - uses: actions/setup-python@v5 + with: + python-version: "3.11" + cache: pip + + - uses: actions/setup-node@v4 + with: + node-version: "20" + + - name: Install qualification dependencies + run: | + python -m pip install --upgrade pip + pip install -e ".[dev]" build twine + + - name: Verify tag matches package version + env: + TARGET_VERSION: ${{ github.event_name == 'workflow_dispatch' && inputs.version || github.ref_name }} + run: | + python - <<'PY' + import os + import tomllib + + expected = os.environ["TARGET_VERSION"].removeprefix("v") + with open("pyproject.toml", "rb") as handle: + actual = tomllib.load(handle)["project"]["version"] + if actual != expected: + raise SystemExit(f"tag/package mismatch: tag={expected} package={actual}") + PY + + - name: Run full credential-free release gate + run: | + python scripts/run_product_qualification.py \ + --output product-qualification.json + + - name: Upload qualification evidence + if: always() + uses: actions/upload-artifact@v4 + with: + name: release-qualification + path: product-qualification.json + if-no-files-found: error + publish: name: Publish to PyPI runs-on: ubuntu-latest + needs: qualify environment: pypi-release permissions: @@ -18,6 +75,8 @@ jobs: steps: - uses: actions/checkout@v4 + with: + ref: ${{ github.event_name == 'workflow_dispatch' && inputs.version || github.ref }} - name: Set up Python 3.11 uses: actions/setup-python@v5 @@ -39,6 +98,27 @@ jobs: - name: Publish to PyPI uses: pypa/gh-action-pypi-publish@release/v1 - # Manual until PyPI trusted publishing is configured for: - # repo=clyrai/SCM_OpenSource, workflow=.github/workflows/release.yml, - # environment=pypi-release. + + github-release: + name: Create GitHub Release + if: startsWith(github.ref, 'refs/tags/') + runs-on: ubuntu-latest + needs: publish + + permissions: + contents: write + + steps: + - uses: actions/checkout@v4 + + - name: Create release notes from CHANGELOG + run: | + VERSION="${GITHUB_REF_NAME#v}" + awk "/^## v${VERSION}/{flag=1; next} /^## /{flag=0} flag" CHANGELOG.md > release_notes.md + + - name: Create GitHub release + uses: softprops/action-gh-release@v2 + with: + body_path: release_notes.md + draft: false + prerelease: false diff --git a/.gitignore b/.gitignore index ca12e70..2813632 100644 --- a/.gitignore +++ b/.gitignore @@ -109,3 +109,4 @@ railway.json research/metrics/ research/reproducibility/ research/benchmarks/ +quality/reports/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 22a34f0..c3e58a5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -19,6 +19,15 @@ Format: each release lists what shipped, why it shipped, what tests verified it, - Documented the optional `scm-memory[llm]` extra required for OpenAI or other OpenAI-compatible providers, while preserving the no-key local first run. - Exposed `scm.__version__` for normal SDK version checks. +- Added checksummed atomic per-user REST/MCP snapshots, restart restoration, + corrupt-state quarantine, same-user operation serialization, bounded queues, + and graceful pool draining on shutdown. +- Tightened the public API with strict request schemas, payload limits, safe + CORS defaults, real OpenAPI paths for the five canonical tools, and encrypted + BYOK storage with legacy-read compatibility. +- Added a credential-free release qualification runner and 46 deterministic + production tests covering abuse inputs, concurrency, durability, recovery, + contract parity, security, and bounded latency. ### Verification @@ -26,6 +35,8 @@ Format: each release lists what shipped, why it shipped, what tests verified it, transition versioning, and SDK schema extraction after sleep. - Re-ran the paper's compact `gpt-5.4-mini` lifecycle scenario against an installed wheel after the behavior fixes. +- Full clean-wheel qualification validates the CLI, demo, REST contract, + server restart durability, JavaScript SDK, dependency audit, and secret scan. --- diff --git a/README.md b/README.md index df8d77b..16ec7b7 100644 --- a/README.md +++ b/README.md @@ -104,6 +104,19 @@ scm mcp Use it with Claude Desktop, Cursor, VS Code, or any MCP-compatible agent. +## Release Qualification + +Before publishing a package, run the credential-free product gate: + +```bash +venv/bin/python scripts/run_product_qualification.py +``` + +It runs the full regression suite, builds and installs the wheel outside the +repository, verifies REST state survives a server restart, and exercises the +live JavaScript SDK. The exact release contract is in +[docs/QUALITY_GATES.md](docs/QUALITY_GATES.md). + ## Optional Provider Quality SCM does not require a cloud key for the local-first path. When you want diff --git a/docs/PUBLISH.md b/docs/PUBLISH.md index 6231e5c..13d7f86 100644 --- a/docs/PUBLISH.md +++ b/docs/PUBLISH.md @@ -1,108 +1,90 @@ -# Publish Runbook +# Publishing SCM -This runbook publishes the Python and JavaScript product packages after CI and -fresh-install smoke tests pass. +SCM publishes only from a clean release commit after the product qualification +report says `ready_for_release: true`. Registry credentials must never be +passed as CLI arguments, committed to the repository, pasted into chat, or +included in screenshots. -## Current Package Targets +## Target Version -| Registry | Package | Target version | +| Registry | Package | Version | |---|---|---| -| PyPI | `scm-memory` | `0.9.1` | -| npm | `scm-memory` | `0.9.1` | +| PyPI | `scm-memory` | `0.9.2` | +| npm | `scm-memory` | `0.9.2` | -Never publish a version until the branch has passed CI, installed-wheel smoke, -REST smoke, JS tests, and tracked-file secret scan. +The following values must match before a tag is created: + +- `pyproject.toml` +- `sdk/js/package.json` +- `src/version.py` +- `CHANGELOG.md` +- Git tag `v0.9.2` + +## Qualification Gate + +```bash +venv/bin/python scripts/run_product_qualification.py \ + --output /tmp/scm-v0.9.2-qualification.json +``` + +Do not publish if this exits nonzero, reports blockers, or skips a required +step. The gate runs the full test suite, builds wheel and sdist artifacts, +installs the wheel in a clean environment, verifies server-restart durability, +exercises the live JavaScript SDK, audits dependencies, and scans tracked +files for secrets. ## PyPI -The GitHub workflow is manual because trusted publishing must match: +The release workflow uses PyPI trusted publishing. Configure PyPI to trust: - repository: `clyrai/SCM_OpenSource` - workflow: `.github/workflows/release.yml` - environment: `pypi-release` -If trusted publishing is configured, run the manual workflow for the target -tag/version. If using local `twine`, keep the token outside the repo. - -Local verification before upload: +After the merged release commit passes qualification and CI: ```bash -rm -rf dist build *.egg-info -python -m build --wheel --sdist -twine check dist/* - -rm -rf /tmp/scm-pypi-smoke -python -m venv /tmp/scm-pypi-smoke -/tmp/scm-pypi-smoke/bin/pip install dist/scm_memory-*.whl -cd /tmp -/tmp/scm-pypi-smoke/bin/scm doctor -/tmp/scm-pypi-smoke/bin/scm demo --dry-run -/tmp/scm-pypi-smoke/bin/python - <<'PY' -from scm import SCMClient, SCMEngine -engine = SCMEngine(session_id="publish-smoke", sandbox=True, offline=True) -assert engine.add_memory("Alice is allergic to peanuts.")["ok"] -assert engine.search_memory("what should Alice avoid?")["ok"] -assert SCMClient(user_id="publish-smoke").user_id == "publish-smoke" -PY +git tag v0.9.2 +git push public v0.9.2 ``` -Publish only after the smoke passes: +The tag workflow reruns qualification, validates tag/version parity, builds +fresh artifacts, runs `twine check`, and publishes only after those gates pass. -```bash -twine upload dist/* -``` - -Verify from the registry: +Verify the registry package from a new environment: ```bash -rm -rf /tmp/scm-registry-smoke -python -m venv /tmp/scm-registry-smoke -/tmp/scm-registry-smoke/bin/pip install scm-memory==0.9.1 -/tmp/scm-registry-smoke/bin/scm doctor +python3 -m venv /tmp/scm-pypi-check +/tmp/scm-pypi-check/bin/pip install scm-memory==0.9.2 +SCM_DATA_DIR=/tmp/scm-pypi-data /tmp/scm-pypi-check/bin/scm doctor +SCM_DATA_DIR=/tmp/scm-pypi-data /tmp/scm-pypi-check/bin/scm demo --dry-run ``` ## npm -First verify the package contents: +Before publishing the JavaScript SDK from the same release commit: ```bash cd sdk/js npm test -npm pack --dry-run -``` - -Publish: - -```bash +npm pack --dry-run --json npm publish --access=public ``` -Verify: +Verify it independently: ```bash -rm -rf /tmp/scm-npm-smoke -mkdir -p /tmp/scm-npm-smoke -cd /tmp/scm-npm-smoke +mkdir -p /tmp/scm-npm-check +cd /tmp/scm-npm-check npm init -y -npm install scm-memory -node --input-type=module -e 'import { SCM } from "scm-memory"; console.log(new SCM({ userId: "smoke" }).userId)' +npm install scm-memory@0.9.2 +node --input-type=module -e \ + "import { SCM } from 'scm-memory'; console.log(new SCM().baseUrl)" ``` -## Release Checklist - -- [ ] Version bumped in `pyproject.toml`. -- [ ] Version bumped in `sdk/js/package.json`. -- [ ] CHANGELOG entry added. -- [ ] `git diff --check` passes. -- [ ] Focused Python tests pass. -- [ ] Wheel and sdist build. -- [ ] Installed-wheel smoke passes outside the repo. -- [ ] REST five-tool smoke passes. -- [ ] JS tests and `npm pack --dry-run` pass. -- [ ] Tracked-file secret scan passes. -- [ ] GitHub release/tag created. -- [ ] PyPI package installed from registry. -- [ ] npm package installed from registry. - -The paper push remains gated until the product front door, install paths, and -demo flow are real. +## Incident Rule + +If a credential appears outside its intended secret store, revoke it before +attempting a release. A published package version cannot be overwritten; yank +or deprecate it, increment the patch version, rerun qualification, and publish +the corrected build. diff --git a/docs/QUALITY_GATES.md b/docs/QUALITY_GATES.md new file mode 100644 index 0000000..289b9db --- /dev/null +++ b/docs/QUALITY_GATES.md @@ -0,0 +1,56 @@ +# Product quality gates + +SCM release readiness is decided by executable gates, not by a raw test count. +The default qualification is offline and strips provider, registry, and other +secret-bearing environment variables before starting child processes. + +## Run the release gate + +```bash +venv/bin/python scripts/run_product_qualification.py +``` + +For a short local iteration loop: + +```bash +venv/bin/python scripts/run_product_qualification.py --fast +``` + +The full gate emits a machine-readable JSON report under `quality/reports/`. +GitHub Actions uploads the same report as a build artifact. + +## Blocking layers + +1. Source integrity: compilation, strict request schemas, closed tool schemas, + version parity, and tracked-file secret scanning. +2. Lifecycle correctness: remember, search, sleep, wake summary, forget, and + contradiction-aware behavior remain callable without a paid provider. +3. Isolation and concurrency: one engine per user under races, serialized + same-user access, parallel different-user access, bounded queues, and no + accepted-write loss during shutdown. +4. Recovery: checksummed atomic snapshots, restart durability, path-safe user + storage, last-known-good preservation, and corrupt-snapshot quarantine. +5. Security: bounded payloads, no credential echo, safe wildcard CORS, + authenticated AES-GCM protection for BYOK values, and tenant isolation. +6. Distribution: wheel and sdist validation, clean-venv install outside the + source tree, packaged resources, `scm doctor`, and offline `scm demo`. +7. Live integration: installed REST server, kill/restart recall, canonical + OpenAPI routes, and a real JavaScript SDK lifecycle against that server. +8. Regression: the complete local pytest suite must exit zero. Skips remain + visible and are not counted as passes. + +## Current budgets + +- Async add acceptance p99: less than 50 ms in the deterministic fake-engine + load test. +- REST health probe p95: less than 250 ms in local ASGI qualification. +- Per-user pending ingest count: bounded by `SCM_MAX_PENDING_PER_USER` + (default 1000), returning HTTP 429 when full. +- Graceful pool shutdown: 30 seconds by default; release tests use explicit + shorter deadlines and require every accepted write to drain. +- Request body: 1 MiB by default through `SCM_MAX_REQUEST_BYTES`. + +These tests qualify the local-first, single-process v0.9.1 runtime. They do not +claim multi-region availability, distributed consensus, or hosted-service SLOs. +Those require deployment infrastructure, traffic replay, fault injection, and +long-running soak evidence beyond this repository. diff --git a/pyproject.toml b/pyproject.toml index 353ae7b..bbf2a04 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,17 +25,24 @@ classifiers = [ "Operating System :: OS Independent", ] dependencies = [ - "fastapi>=0.100.0", + "fastapi>=0.139.0", "uvicorn>=0.23.0", "pydantic>=2.0", "networkx>=3.0", "numpy>=1.24.0", "python-dotenv>=1.0.0", - "requests>=2.28.0", + "tomli>=2.0; python_version < '3.11'", + "requests>=2.34.2", + "urllib3>=2.7.0", + "idna>=3.15", "sqlalchemy>=2.0", "rich>=13.0", "ollama>=0.1.0", - "mcp>=1.0.0", + "mcp>=1.28.1", + "cryptography>=48.0.1", + "pydantic-settings>=2.14.2", + "pyjwt>=2.13.0", + "python-multipart>=0.0.31", ] [project.optional-dependencies] @@ -44,7 +51,27 @@ llm = ["openai>=1.0.0"] postgres = ["psycopg2-binary>=2.9", "pgvector>=0.3.0"] langchain = ["langchain-core>=0.2.0", "langchain>=0.2.0"] all = ["sentence-transformers>=2.2.0", "openai>=1.0.0", "ollama>=0.1.0", "psycopg2-binary>=2.9", "pgvector>=0.3.0", "langchain-core>=0.2.0", "langchain>=0.2.0"] -dev = ["pytest>=7.0", "pytest-asyncio>=0.21", "jsonschema>=4.0"] +dev = [ + "pytest>=7.0", + "pytest-asyncio>=0.21", + "pytest-cov>=5.0", + "pytest-timeout>=2.3", + "hypothesis>=6.100", + "jsonschema>=4.0", + "pip-audit>=2.7", + "ruff>=0.6", + "bandit>=1.7", +] + +[tool.pytest.ini_options] +addopts = "--strict-markers" +markers = [ + "production: deterministic product release qualification", + "load: bounded performance and concurrency qualification", + "security: security and secret-handling qualification", + "recovery: durability and crash-recovery qualification", +] +timeout = 120 [project.scripts] scm = "src.cli.main:main" diff --git a/scm/__init__.py b/scm/__init__.py index af8752b..7ef7578 100644 --- a/scm/__init__.py +++ b/scm/__init__.py @@ -1,16 +1,8 @@ -""" -SCM product SDK. -""" - -from importlib.metadata import PackageNotFoundError, version +"""Public SCM product SDK.""" from src.integrations.langchain_adapter import SCMClient +from src.version import __version__ from .runtime import SCMEngine, list_profiles -try: - __version__ = version("scm-memory") -except PackageNotFoundError: # source checkout, before a wheel is installed - __version__ = "0.9.2" - __all__ = ["SCMClient", "SCMEngine", "list_profiles", "__version__"] diff --git a/scripts/installed_wheel_smoke.py b/scripts/installed_wheel_smoke.py new file mode 100644 index 0000000..66d2db9 --- /dev/null +++ b/scripts/installed_wheel_smoke.py @@ -0,0 +1,154 @@ +"""Smoke an installed SCM wheel from outside its source checkout.""" +from __future__ import annotations + +import argparse +import importlib.resources as resources +import json +import os +from pathlib import Path +import socket +import subprocess +import sys +import tempfile +import time +from urllib import request + + +def _free_port() -> int: + with socket.socket() as sock: + sock.bind(("127.0.0.1", 0)) + return int(sock.getsockname()[1]) + + +def _request(base: str, method: str, path: str, payload=None): + data = None + headers = {} + if payload is not None: + data = json.dumps(payload).encode("utf-8") + headers["Content-Type"] = "application/json" + req = request.Request(base + path, data=data, headers=headers, method=method) + with request.urlopen(req, timeout=20) as response: + return json.loads(response.read().decode("utf-8")) + + +def _wait_for_health(base: str, process: subprocess.Popen) -> None: + for _ in range(120): + if process.poll() is not None: + raise RuntimeError(f"SCM server exited early with {process.returncode}") + try: + if _request(base, "GET", "/health").get("ok"): + return + except Exception: + time.sleep(0.25) + raise RuntimeError("SCM server did not become healthy") + + +def _stop(process: subprocess.Popen) -> None: + if process.poll() is not None: + return + process.terminate() + try: + process.wait(timeout=30) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=10) + + +def _start_server(port: int, log_handle) -> subprocess.Popen: + executable = Path(sys.executable).parent / "scm" + return subprocess.Popen( + [str(executable), "serve", "--host", "127.0.0.1", "--port", str(port)], + cwd=Path.cwd(), + env=os.environ.copy(), + stdout=log_handle, + stderr=subprocess.STDOUT, + text=True, + ) + + +def _assert_installed_import(expected_version: str, repo_root: Path) -> None: + import scm + from scm import SCMClient, SCMEngine, __version__ + + assert __version__ == expected_version + assert SCMEngine.__name__ == "SCMEngine" + assert SCMClient.__name__ == "SCMClient" + assert repo_root not in Path(scm.__file__).resolve().parents + assert resources.files("src.core").joinpath("locales/en.json").is_file() + assert resources.files("src.api").joinpath("static/demo.html").is_file() + + engine = SCMEngine(session_id="wheel-sdk", sandbox=True, offline=True) + added = engine.add_memory("Mira is allergic to shellfish.") + assert added["ok"] and added["memory_id"] + assert engine.search_memory("What should Mira avoid?")["ok"] + assert engine.sleep("micro")["ok"] + assert engine.wake_summary()["ok"] + assert engine.forget(added["memory_id"])["ok"] + + +def _http_restart_and_js_smoke(js_smoke: Path) -> None: + port = _free_port() + base = f"http://127.0.0.1:{port}/v1" + user_id = "installed-wheel-restart" + sentinel = "The restart sentinel is heliotrope 4821." + + with tempfile.NamedTemporaryFile(prefix="scm-wheel-server-", suffix=".log") as log: + first = _start_server(port, log) + try: + _wait_for_health(base, first) + tools = _request(base, "GET", "/tools?format=openai")["tools"] + assert [tool["function"]["name"] for tool in tools] == [ + "add_memory", "search_memory", "sleep", "wake_summary", "forget" + ] + added = _request( + base, + "POST", + "/memories", + {"user_id": user_id, "text": sentinel, "sync": True}, + ) + assert added["ok"] + assert _request( + base, + "POST", + "/memories/search", + {"user_id": user_id, "query": "restart sentinel", "wait_for_pending": True}, + )["ok"] + finally: + _stop(first) + + second = _start_server(port, log) + try: + _wait_for_health(base, second) + found = _request( + base, + "POST", + "/memories/search", + {"user_id": user_id, "query": "restart sentinel", "wait_for_pending": True}, + ) + assert "heliotrope 4821" in json.dumps(found).lower() + subprocess.run( + ["node", str(js_smoke), base], + cwd=Path.cwd(), + env=os.environ.copy(), + check=True, + timeout=120, + ) + finally: + _stop(second) + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--expected-version", required=True) + parser.add_argument("--repo-root", type=Path, required=True) + parser.add_argument("--js-smoke", type=Path, required=True) + args = parser.parse_args() + + _assert_installed_import(args.expected_version, args.repo_root.resolve()) + _http_restart_and_js_smoke(args.js_smoke.resolve()) + print(json.dumps({"ok": True, "version": args.expected_version})) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/run_product_qualification.py b/scripts/run_product_qualification.py new file mode 100644 index 0000000..aec9cad --- /dev/null +++ b/scripts/run_product_qualification.py @@ -0,0 +1,275 @@ +"""Run SCM's credential-free product release qualification and emit JSON.""" +from __future__ import annotations + +import argparse +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +import json +import os +from pathlib import Path +import re +import subprocess +import sys +import tempfile +import time +try: + import tomllib +except ModuleNotFoundError: # Python 3.10 + import tomli as tomllib +from typing import List, Optional + + +ROOT = Path(__file__).resolve().parents[1] +SECRET_ENV = re.compile(r"(?:API[_-]?KEY|TOKEN|SECRET|PASSWORD|TWINE|NPM_AUTH)", re.I) +SECRET_TEXT = [ + re.compile(r"sk-(?:proj-)?[A-Za-z0-9_-]{20,}"), + re.compile(r"pypi-[A-Za-z0-9_-]{30,}"), + re.compile(r"npm_[A-Za-z0-9]{20,}"), + re.compile(r"gh[pousr]_[A-Za-z0-9]{20,}"), +] + + +@dataclass +class StepResult: + name: str + required: bool + status: str + return_code: Optional[int] + duration_seconds: float + command: List[str] + output_tail: str + + +def _redact(text: str) -> str: + for pattern in SECRET_TEXT: + text = pattern.sub("[REDACTED]", text) + return text + + +def _offline_env(data_dir: Path) -> dict: + env = { + key: value + for key, value in os.environ.items() + if not SECRET_ENV.search(key) + } + env.pop("PYTHONPATH", None) + env.update( + { + "SCM_DATA_DIR": str(data_dir), + "SCM_API_PERSISTENCE": "1", + "SCM_EMBEDDING_BACKEND": "hash", + "SCM_AUTO_SLEEP_DISABLE": "1", + "IDLE_LEARNER_ENABLED": "false", + "LLM_PROVIDER": "", + "PYTHONHASHSEED": "0", + } + ) + return env + + +def _run( + results: List[StepResult], + name: str, + command: List[str], + *, + env: dict, + cwd: Path = ROOT, + timeout: int = 600, + required: bool = True, +) -> StepResult: + started = time.monotonic() + try: + completed = subprocess.run( + command, + cwd=cwd, + env=env, + text=True, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + timeout=timeout, + check=False, + ) + status = "passed" if completed.returncode == 0 else "failed" + code = completed.returncode + output = completed.stdout or "" + except subprocess.TimeoutExpired as exc: + status = "failed" + code = None + output = f"timed out after {timeout}s\n{exc.stdout or ''}" + duration = time.monotonic() - started + result = StepResult( + name=name, + required=required, + status=status, + return_code=code, + duration_seconds=round(duration, 3), + command=command, + output_tail=_redact(output[-12_000:]), + ) + results.append(result) + state = "PASS" if status == "passed" else "FAIL" + print(f"[{state}] {name} ({duration:.2f}s)", flush=True) + return result + + +def _skipped(results: List[StepResult], name: str, required: bool = False) -> None: + results.append(StepResult(name, required, "skipped", None, 0.0, [], "")) + print(f"[SKIP] {name}", flush=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--fast", action="store_true", help="skip full regression and wheel install") + parser.add_argument("--output", type=Path) + args = parser.parse_args() + + version = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))["project"]["version"] + timestamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + output = args.output or ROOT / "quality" / "reports" / f"qualification-{timestamp}.json" + results: List[StepResult] = [] + started_at = datetime.now(timezone.utc) + + with tempfile.TemporaryDirectory(prefix="scm-qualification-") as temp_name: + temp = Path(temp_name) + env = _offline_env(temp / "source-data") + python = sys.executable + + _run(results, "compile", [python, "-m", "compileall", "-q", "scm", "src", "tests/production"], env=env) + _run( + results, + "critical-static-analysis", + [ + python, "-m", "ruff", "check", + "--select", "E9,F63,F7,F82", + "scm", "src", "tests/production", "scripts", + ], + env=env, + ) + _run( + results, + "high-severity-security-analysis", + [ + python, "-m", "bandit", "-q", "-lll", "-r", + "scm", "src/api", "src/cloud", "src/integrations", + ], + env=env, + timeout=300, + ) + _run(results, "production-tests", [python, "-m", "pytest", "tests/production", "-q", "--tb=short"], env=env, timeout=300) + _run(results, "js-unit-tests", ["npm", "test"], cwd=ROOT / "sdk/js", env=env, timeout=120) + _run(results, "js-package-dry-run", ["npm", "pack", "--dry-run", "--json"], cwd=ROOT / "sdk/js", env=env, timeout=120) + + if args.fast: + _skipped(results, "full-regression") + _skipped(results, "clean-wheel-install") + _skipped(results, "dependency-audit") + else: + _run(results, "full-regression", [python, "-m", "pytest", "tests", "-q", "--tb=short"], env=env, timeout=900) + dist = temp / "dist" + build = _run( + results, + "build-distributions", + [python, "-m", "build", "--wheel", "--sdist", "--outdir", str(dist)], + env=env, + timeout=300, + ) + if build.status == "passed": + _run(results, "twine-check", [python, "-m", "twine", "check", *map(str, sorted(dist.iterdir()))], env=env) + wheel = next(dist.glob("*.whl")) + clean_venv = temp / "venv" + _run(results, "create-clean-venv", [python, "-m", "venv", str(clean_venv)], env=env) + clean_python = clean_venv / "bin" / "python" + clean_scm = clean_venv / "bin" / "scm" + _run( + results, + "upgrade-clean-pip", + [str(clean_python), "-m", "pip", "install", "--upgrade", "pip"], + env=env, + cwd=temp, + timeout=300, + ) + install = _run( + results, + "install-clean-wheel", + [str(clean_python), "-m", "pip", "install", str(wheel)], + env=env, + cwd=temp, + timeout=600, + ) + if install.status == "passed": + wheel_env = _offline_env(temp / "wheel-data") + _run(results, "wheel-pip-check", [str(clean_python), "-m", "pip", "check"], env=wheel_env, cwd=temp) + _run(results, "wheel-cli-help", [str(clean_scm), "--help"], env=wheel_env, cwd=temp) + _run(results, "wheel-doctor", [str(clean_scm), "doctor", "--json"], env=wheel_env, cwd=temp, timeout=120) + _run(results, "wheel-offline-demo", [str(clean_scm), "demo", "--dry-run"], env=wheel_env, cwd=temp, timeout=180) + _run( + results, + "wheel-sdk-rest-restart-js", + [ + str(clean_python), + str(ROOT / "scripts/installed_wheel_smoke.py"), + "--expected-version", version, + "--repo-root", str(ROOT), + "--js-smoke", str(ROOT / "sdk/js/test/live-smoke.mjs"), + ], + env=wheel_env, + cwd=temp, + timeout=300, + ) + frozen = subprocess.check_output( + [str(clean_python), "-m", "pip", "freeze", "--all"], + cwd=temp, + env=wheel_env, + text=True, + ) + audit_requirements = temp / "audit-requirements.txt" + audit_requirements.write_text( + "\n".join( + line + for line in frozen.splitlines() + if not line.lower().replace("_", "-").startswith("scm-memory") + ) + + "\n", + encoding="utf-8", + ) + _run( + results, + "dependency-audit", + [ + python, + "-m", + "pip_audit", + "-r", + str(audit_requirements), + "--no-deps", + "--progress-spinner=off", + "--strict", + ], + env=env, + timeout=300, + ) + else: + _skipped(results, "twine-check", required=True) + _skipped(results, "clean-wheel-install", required=True) + + blockers = [result.name for result in results if result.required and result.status != "passed"] + report = { + "schema_version": 1, + "product": "scm-memory", + "version": version, + "started_at": started_at.isoformat(), + "finished_at": datetime.now(timezone.utc).isoformat(), + "credential_policy": "provider and publishing credentials stripped", + "ready_for_release": not blockers, + "blockers": blockers, + "steps": [asdict(result) for result in results], + } + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(report, indent=2) + "\n", encoding="utf-8") + print(f"Report: {output}") + print(f"Release ready: {report['ready_for_release']}") + return 0 if report["ready_for_release"] else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sdk/js/package.json b/sdk/js/package.json index 640ab61..2ce8513 100644 --- a/sdk/js/package.json +++ b/sdk/js/package.json @@ -1,6 +1,6 @@ { "name": "scm-memory", - "version": "0.9.1", + "version": "0.9.2", "description": "SCM lifecycle memory client for any agent: add, search, sleep, wake summary, forget.", "main": "src/index.js", "types": "src/index.d.ts", diff --git a/sdk/js/test/live-smoke.mjs b/sdk/js/test/live-smoke.mjs new file mode 100644 index 0000000..0b69cd3 --- /dev/null +++ b/sdk/js/test/live-smoke.mjs @@ -0,0 +1,36 @@ +import assert from "node:assert/strict"; + +import { SCM } from "../src/index.js"; + +const baseUrl = process.argv[2]; +assert.ok(baseUrl, "usage: node live-smoke.mjs "); + +const scm = new SCM({ + baseUrl, + userId: `js-live-${process.pid}`, + timeoutMs: 30_000, +}); + +const health = await scm.health(); +assert.equal(health.ok, true); + +const listed = await scm.listTools("openai"); +assert.deepEqual( + listed.tools.map((tool) => tool.function.name), + ["add_memory", "search_memory", "sleep", "wake_summary", "forget"], +); + +const sentinel = "JavaScript live sentinel is indigo 8842."; +assert.equal((await scm.addMemory(sentinel)).ok, true); +assert.equal((await scm.sleep("micro")).ok, true); + +const found = await scm.searchMemory("JavaScript live sentinel", 5); +assert.equal(found.ok, true); +assert.match(JSON.stringify(found).toLowerCase(), /indigo 8842/); + +const memory = found.memories.find((item) => item.description.includes("indigo 8842")); +assert.ok(memory?.id); +assert.equal((await scm.wakeSummary(24)).ok, true); +assert.equal((await scm.forget(memory.id)).ok, true); + +console.log(JSON.stringify({ ok: true, tools: 5, lifecycle: "complete" })); diff --git a/src/api/demo_router.py b/src/api/demo_router.py index d725cac..c982cc9 100644 --- a/src/api/demo_router.py +++ b/src/api/demo_router.py @@ -56,6 +56,7 @@ _search_memory_handler, _wake_summary_handler, ) +from ..lifecycle.wake_summary import WakeSummaryBuilder from ..runtime_factory import build_lifecycle_engine router = APIRouter(prefix="/demo", tags=["demo"]) diff --git a/src/api/main.py b/src/api/main.py index 344dd3a..32aa79a 100644 --- a/src/api/main.py +++ b/src/api/main.py @@ -3,13 +3,14 @@ from fastapi import Request, Response from fastapi.middleware.cors import CORSMiddleware from fastapi.staticfiles import StaticFiles -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, JSONResponse from contextlib import asynccontextmanager import uvicorn import os from time import monotonic from .memory import router as memory_router, init_memory_components +from ..integrations.memories_api import close_pool as close_memories_pool from ..sleep.api import router as sleep_router from ..api.chat import router as chat_router from ..api import chat as chat_module @@ -35,6 +36,7 @@ observe_http_request, render_metrics_payload, ) +from ..version import __version__ # Process-global so other modules (e.g. the future wake-summary endpoint) can @@ -94,21 +96,28 @@ async def lifespan(app: FastAPI): idle_learner.stop() chat_module._idle_learner = None print("IdleLearner stopped cleanly.") + if not close_memories_pool(timeout=30.0): + print("SCM memory pool shutdown exceeded its drain deadline.") app = FastAPI( title="SCM", description="Local-first lifecycle memory runtime with sleep consolidation", - version="0.9.2", + version=__version__, lifespan=lifespan ) LOGGER = get_structured_logger("scm.api.main") -# CORS middleware +_cors_origins = [ + item.strip() + for item in os.environ.get("SCM_CORS_ORIGINS", "*").split(",") + if item.strip() +] or ["*"] + app.add_middleware( CORSMiddleware, - allow_origins=["*"], - allow_credentials=True, + allow_origins=_cors_origins, + allow_credentials="*" not in _cors_origins, allow_methods=["*"], allow_headers=["*"], ) @@ -157,6 +166,19 @@ async def lifespan(app: FastAPI): @app.middleware("http") async def observability_middleware(request: Request, call_next): + max_request_bytes = int(os.environ.get("SCM_MAX_REQUEST_BYTES", "1048576")) + content_length = request.headers.get("content-length") + if content_length: + try: + if int(content_length) > max_request_bytes: + return JSONResponse( + {"error": "request body too large", "max_bytes": max_request_bytes}, + status_code=413, + ) + except ValueError: + return JSONResponse( + {"error": "invalid Content-Length header"}, status_code=400 + ) start = monotonic() route = request.scope.get("route") path = getattr(route, "path", request.url.path) @@ -219,7 +241,7 @@ async def root(): return FileResponse(landing_path, media_type="text/html") return { "name": "SCM", - "version": "0.9.2", + "version": __version__, "status": "running", "description": "Memory that works like yours — wake + sleep phases for AI agents", } diff --git a/src/cli/main.py b/src/cli/main.py index 2fe0bec..16a0517 100644 --- a/src/cli/main.py +++ b/src/cli/main.py @@ -247,17 +247,9 @@ def _apply_demo_defaults() -> str: def _version_string() -> str: - try: - import importlib.metadata as md - return md.version("scm-memory") - except Exception: - try: - import tomllib - here = Path(__file__).resolve().parents[2] - data = tomllib.loads((here / "pyproject.toml").read_text()) - return str(data["project"]["version"]) - except Exception: - return "(unknown)" + from src.version import __version__ + + return __version__ def _redacted_env_value(name: str) -> str: @@ -375,13 +367,13 @@ def check_provider_configuration() -> str: if provider not in {"ollama", "deepseek", "openai"}: raise RuntimeError(f"unsupported LLM_PROVIDER={provider!r}") if provider in {"deepseek", "openai"}: + key_env = "OPENAI_API_KEY" if provider == "openai" else "DEEPSEEK_API_KEY" + if not os.environ.get(key_env): + raise RuntimeError(f"{key_env} is required for LLM_PROVIDER={provider}") import importlib.util if importlib.util.find_spec("openai") is None: raise RuntimeError("install the provider extra: pip install 'scm-memory[llm]'") - key_env = "OPENAI_API_KEY" if provider == "openai" else "DEEPSEEK_API_KEY" - if not os.environ.get(key_env): - raise RuntimeError(f"{key_env} is required for LLM_PROVIDER={provider}") return f"{provider} configured" checks.append(("provider_configuration", *_doctor_check("provider_configuration", check_provider_configuration))) @@ -393,6 +385,11 @@ def check_provider_configuration() -> str: "SCM_EMBEDDING_BACKEND": _redacted_env_value("SCM_EMBEDDING_BACKEND"), "SCM_DATA_DIR": _redacted_env_value("SCM_DATA_DIR"), } + providers = { + "openai": bool(os.environ.get("OPENAI_API_KEY")), + "deepseek": bool(os.environ.get("DEEPSEEK_API_KEY")), + "ollama": bool(os.environ.get("OLLAMA_BASE_URL")), + } ok = all(item[1] for item in checks) if args.json: print(json.dumps({ @@ -402,6 +399,7 @@ def check_provider_configuration() -> str: for name, passed, detail in checks ], "environment": env, + "providers": providers, }, indent=2)) return 0 if ok else 1 @@ -412,6 +410,8 @@ def check_provider_configuration() -> str: print(" env provider config:") for key, value in env.items(): print(f" {key}: {value}") + present = [name for name, configured in providers.items() if configured] + print(f" providers: {', '.join(present) if present else 'none (offline ready)'}") if ok: print() print("SCM is ready for local plug-and-play use.") diff --git a/src/cloud/accounts.py b/src/cloud/accounts.py index 9860158..5fccfa0 100644 --- a/src/cloud/accounts.py +++ b/src/cloud/accounts.py @@ -21,27 +21,20 @@ import hmac import os import secrets +import threading import uuid from base64 import urlsafe_b64decode, urlsafe_b64encode from datetime import datetime, timezone from typing import Any, Dict, List, Optional +from cryptography.hazmat.primitives.ciphers.aead import AESGCM -# ── Symmetric encryption for at-rest BYOK API keys ───────────────────── +# ── Authenticated encryption for at-rest BYOK API keys ──────────────── -def _xor_bytes(data: bytes, key: bytes) -> bytes: - """Stream-cipher XOR (one-time-pad-style with derived stream). - - Not AES, but sufficient for at-rest protection of LLM API keys against - db dump leaks. Cloud production should swap this for AES-GCM via - cryptography lib; for the bootstrap phase we keep zero new - dependencies and use a SHA256-derived keystream. - The key is derived from SCM_CLOUD_SECRET_KEY so the deployment can - rotate it; rotating invalidates existing encrypted BYOK keys (they - must be re-set), which is the desired behavior. - """ +def _xor_bytes(data: bytes, key: bytes) -> bytes: + """Decrypt legacy pre-v0.9.1 ciphertext during migration only.""" out = bytearray(len(data)) pos = 0 counter = 0 @@ -77,13 +70,31 @@ def _master_key() -> bytes: def _encrypt(plaintext: str) -> str: if not plaintext: return "" - blob = _xor_bytes(plaintext.encode("utf-8"), _master_key()) - return urlsafe_b64encode(blob).decode("ascii") + nonce = secrets.token_bytes(12) + ciphertext = AESGCM(_master_key()).encrypt( + nonce, + plaintext.encode("utf-8"), + b"scm-byok-v1", + ) + return "v1:" + urlsafe_b64encode(nonce + ciphertext).decode("ascii") def _decrypt(ciphertext: str) -> str: if not ciphertext: return "" + if ciphertext.startswith("v1:"): + blob = urlsafe_b64decode(ciphertext[3:].encode("ascii")) + if len(blob) < 29: + raise ValueError("invalid BYOK ciphertext") + nonce, encrypted = blob[:12], blob[12:] + plaintext = AESGCM(_master_key()).decrypt( + nonce, + encrypted, + b"scm-byok-v1", + ) + return plaintext.decode("utf-8") + + # Backward-compatible read path. The next set/rotation writes AES-GCM. blob = urlsafe_b64decode(ciphertext.encode("ascii")) return _xor_bytes(blob, _master_key()).decode("utf-8", errors="replace") @@ -119,6 +130,7 @@ def _now() -> str: _SCHEMA_ENSURED = False +_SCHEMA_LOCK = threading.Lock() def _conn(): @@ -129,8 +141,10 @@ def _conn(): global _SCHEMA_ENSURED from ..core.sqlite_db import get_connection, init_db if not _SCHEMA_ENSURED: - init_db() - _SCHEMA_ENSURED = True + with _SCHEMA_LOCK: + if not _SCHEMA_ENSURED: + init_db() + _SCHEMA_ENSURED = True return get_connection() diff --git a/src/core/sqlite_db.py b/src/core/sqlite_db.py index 5c77af9..99ec5a9 100644 --- a/src/core/sqlite_db.py +++ b/src/core/sqlite_db.py @@ -6,12 +6,15 @@ from typing import List, Optional, Dict, Any import json import os +import threading from ..core.config import DATA_DIR from .time_utils import ensure_utc, utc_isoformat DATA_DIR.mkdir(parents=True, exist_ok=True) DB_PATH = DATA_DIR / "sleepai.db" +_DB_INIT_LOCK = threading.RLock() +_MEMORY_LOCK = threading.Lock() def set_db_path(path) -> None: @@ -31,15 +34,23 @@ def set_db_path(path) -> None: def get_connection(): """Get SQLite connection""" - conn = sqlite3.connect(str(DB_PATH)) + conn = sqlite3.connect(str(DB_PATH), timeout=30.0) conn.row_factory = sqlite3.Row + conn.execute("PRAGMA busy_timeout = 30000") + conn.execute("PRAGMA foreign_keys = ON") return conn def init_db(): + with _DB_INIT_LOCK: + _init_db_unlocked() + + +def _init_db_unlocked(): """Initialize SQLite database schema""" conn = get_connection() cursor = conn.cursor() + cursor.execute("PRAGMA journal_mode = WAL") # Concepts table cursor.execute(""" @@ -922,5 +933,7 @@ def list_user_sleep_configs(self) -> List[Dict[str, Any]]: def get_memory() -> SQLiteMemory: global _memory if _memory is None: - _memory = SQLiteMemory() + with _MEMORY_LOCK: + if _memory is None: + _memory = SQLiteMemory() return _memory diff --git a/src/integrations/mcp_server.py b/src/integrations/mcp_server.py index 539e6f6..ae8ac56 100644 --- a/src/integrations/mcp_server.py +++ b/src/integrations/mcp_server.py @@ -32,6 +32,7 @@ from __future__ import annotations import asyncio +from contextlib import contextmanager import json import logging import os @@ -40,13 +41,18 @@ import time from datetime import datetime, timezone from pathlib import Path -from typing import Any, Dict, Optional +from typing import Any, Callable, Dict, Iterator, Optional from .task_context import TaskContextState +from .user_state_store import SnapshotWriteError, UserStateStore logger = logging.getLogger("scm.mcp") +class IngestQueueFull(RuntimeError): + """Raised when a user exceeds the bounded async-ingest queue.""" + + # ─── Per-user engine pool with auto-sleep ───────────────────────────────── @@ -67,15 +73,31 @@ def __init__( sweep_interval_sec: float = 30.0, auto_sleep: bool = True, legacy_idle_mode: Optional[bool] = None, + persistence: Optional[bool] = None, + state_store: Optional[UserStateStore] = None, + max_pending_per_user: Optional[int] = None, ): self._engines: Dict[str, Any] = {} self._last_activity: Dict[str, float] = {} self._cached_summaries: Dict[str, Any] = {} self._task_context: Dict[str, TaskContextState] = {} - self._sleep_lock = threading.Lock() + self._pool_lock = threading.RLock() + self._operation_locks: Dict[str, threading.RLock] = {} + self._accepting = True + self._load_status: Dict[str, str] = {} self._idle_threshold = idle_threshold_sec self._sweep_interval = sweep_interval_sec self._auto_sleep = auto_sleep + if persistence is None: + persistence = os.environ.get("SCM_API_PERSISTENCE", "1") != "0" + self._state_store = state_store or UserStateStore(enabled=bool(persistence)) + if max_pending_per_user is None: + max_pending_per_user = int( + os.environ.get("SCM_MAX_PENDING_PER_USER", "1000") + ) + if max_pending_per_user < 1: + raise ValueError("max_pending_per_user must be >= 1") + self._max_pending_per_user = int(max_pending_per_user) # Backwards compat: when a deployment sets SCM_IDLE_THRESHOLD_SEC # without ever writing a per-user sleep config, keep the legacy # idle-timer behavior. Once the user POSTs a sleep-config, the @@ -100,6 +122,16 @@ def __init__( # `wait_for_pending` can wake immediately instead of polling. self._pending_done = threading.Condition(self._pending_lock) + def _operation_lock_for(self, user_id: str) -> threading.RLock: + with self._pool_lock: + return self._operation_locks.setdefault(user_id, threading.RLock()) + + @contextmanager + def user_operation(self, user_id: str) -> Iterator[None]: + """Serialize mutations and reads for one engine without blocking others.""" + with self._operation_lock_for(user_id): + yield + def get_or_create(self, user_id: str, bump_activity: bool = True) -> Any: """Return the per-user ChatEngine, building it lazily. @@ -111,20 +143,75 @@ def get_or_create(self, user_id: str, bump_activity: bool = True) -> Any: operator manually consolidating doesn't mean the user came back. """ - if user_id in self._engines: - if bump_activity: + user_id = str(user_id or "default") + with self.user_operation(user_id): + with self._pool_lock: + engine = self._engines.get(user_id) + if engine is None: + engine = self._build_engine(user_id) + load_result = self._state_store.load(user_id, engine) + with self._pool_lock: + self._engines[user_id] = engine + self._load_status[user_id] = load_result.status + logger.info( + "[scm.mcp] created engine for user_key=%s snapshot=%s", + self._state_store.user_key(user_id)[:12], + load_result.status, + ) self._touch(user_id) - return self._engines[user_id] - engine = self._build_engine(user_id) - self._engines[user_id] = engine - # Always bump on first creation — the engine wouldn't exist if no - # user activity had triggered it. - self._touch(user_id) - logger.info(f"[scm.mcp] created engine for user {user_id!r}") - return engine + elif bump_activity: + self._touch(user_id) + return engine def _touch(self, user_id: str) -> None: - self._last_activity[user_id] = time.time() + with self._pool_lock: + self._last_activity[user_id] = time.time() + + def persist_user(self, user_id: str, engine: Optional[Any] = None) -> bool: + if not self._state_store.enabled: + return False + with self.user_operation(user_id): + if engine is None: + with self._pool_lock: + engine = self._engines.get(user_id) + if engine is None: + return False + self._state_store.save(user_id, engine) + return True + + def health_snapshot(self) -> Dict[str, Any]: + with self._pool_lock: + active_users = len(self._engines) + load_status = dict(self._load_status) + accepting = self._accepting + with self._pending_lock: + pending = sum(self._pending_count.values()) + degraded = sum(1 for status in load_status.values() if status == "corrupt") + return { + "active_users": active_users, + "pending_ingests": pending, + "accepting": accepting, + "persistence": self._state_store.enabled, + "degraded_snapshots": degraded, + "status": "degraded" if degraded else "healthy", + } + + def call_handler( + self, + tool_name: str, + args: Dict[str, Any], + handler: Callable[[Dict[str, Any], Any], Dict[str, Any]], + *, + bump_activity: bool, + ) -> Dict[str, Any]: + """Run one canonical tool against a race-free per-user engine.""" + user_id = str(args.get("user_id") or "default") + with self.user_operation(user_id): + engine = self.get_or_create(user_id, bump_activity=bump_activity) + result = handler(args, engine) + if tool_name in {"add_memory", "forget"} and result.get("ok"): + self.persist_user(user_id, engine) + return result # ── Async ingest queue ────────────────────────────────────────────────── @@ -143,29 +230,50 @@ def enqueue_ingest( import queue as _queue import uuid - # Lazy-init the per-user queue + worker on first ingest. - if user_id not in self._ingest_queues: - q = _queue.Queue() - self._ingest_queues[user_id] = q - t = threading.Thread( - target=self._ingest_worker, - args=(user_id, q), - name=f"scm-ingest-{user_id}", - daemon=True, - ) - t.start() - self._ingest_workers[user_id] = t + user_id = str(user_id or "default") + with self._pool_lock: + if not self._accepting: + raise RuntimeError("SCM engine pool is shutting down") + # Lazy-init the per-user queue + worker on first ingest. + if user_id not in self._ingest_queues: + q = _queue.Queue(maxsize=self._max_pending_per_user) + self._ingest_queues[user_id] = q + worker_name = self._state_store.user_key(user_id)[:12] + t = threading.Thread( + target=self._ingest_worker, + args=(user_id, q), + name=f"scm-ingest-{worker_name}", + daemon=True, + ) + t.start() + self._ingest_workers[user_id] = t + q = self._ingest_queues[user_id] # Reserve a placeholder id; real concept id is assigned by SCM. placeholder = f"pending_{uuid.uuid4().hex[:12]}" with self._pending_lock: + pending = self._pending_count.get(user_id, 0) + if pending >= self._max_pending_per_user: + raise IngestQueueFull( + f"pending ingest limit reached ({self._max_pending_per_user})" + ) self._pending_count[user_id] = self._pending_count.get(user_id, 0) + 1 - self._ingest_queues[user_id].put({ - "text": text, - "metadata": metadata or {}, - "placeholder": placeholder, - "enqueued_at": time.time(), - }) + try: + q.put_nowait({ + "text": text, + "metadata": metadata or {}, + "placeholder": placeholder, + "enqueued_at": time.time(), + }) + except _queue.Full as exc: + with self._pending_lock: + self._pending_count[user_id] = max( + 0, self._pending_count.get(user_id, 1) - 1 + ) + self._pending_done.notify_all() + raise IngestQueueFull( + f"pending ingest limit reached ({self._max_pending_per_user})" + ) from exc # Bump activity — async ingest IS user activity. self._touch(user_id) return placeholder @@ -173,24 +281,28 @@ def enqueue_ingest( def _ingest_worker(self, user_id: str, q) -> None: """Per-user background worker that drains the ingest queue.""" import queue as _queue - while not self._stop_flag.is_set(): + while True: try: task = q.get(timeout=1.0) except _queue.Empty: + if self._stop_flag.is_set(): + break continue if task is None: # shutdown sentinel + q.task_done() break try: - # Lazy-create engine on first ingest if not already. - engine = self._engines.get(user_id) - if engine is None: - engine = self._build_engine(user_id) - self._engines[user_id] = engine - logger.info(f"[scm.mcp] created engine for user {user_id!r} (via async ingest)") - # The actual ingest — this is what was blocking the API path. - engine.chat(task["text"]) + with self.user_operation(user_id): + engine = self.get_or_create(user_id, bump_activity=False) + # The actual ingest — this is what was blocking the API path. + engine.chat(task["text"]) + self.persist_user(user_id, engine) except Exception as e: - logger.warning(f"[scm.mcp] async ingest failed for {user_id!r}: {e!r}") + logger.warning( + "[scm.mcp] async ingest failed for user_key=%s: %s", + self._state_store.user_key(user_id)[:12], + type(e).__name__, + ) finally: q.task_done() with self._pending_lock: @@ -225,10 +337,11 @@ def fire_sleep_now(self, user_id: str, mode: str = "deep") -> Dict[str, Any]: Used by the manual sleep path so a programmatic sleep produces a wake-summary the same way the autonomous sweeper would. """ - engine = self._engines.get(user_id) - if engine is None: - return {"ok": False, "error": f"no engine for user {user_id!r}"} - with self._sleep_lock: + with self.user_operation(user_id): + with self._pool_lock: + engine = self._engines.get(user_id) + if engine is None: + return {"ok": False, "error": "no engine for user"} try: stats = engine.force_sleep(mode) or {} except Exception as e: @@ -237,17 +350,19 @@ def fire_sleep_now(self, user_id: str, mode: str = "deep") -> Dict[str, Any]: # the right window for a manual cycle. try: from datetime import timedelta - last_act = self._last_activity.get(user_id, time.time()) + with self._pool_lock: + last_act = self._last_activity.get(user_id, time.time()) idle_for = max(60.0, time.time() - last_act + 60.0) since = datetime.now(timezone.utc) - timedelta(seconds=idle_for) summary = engine._wake_summary_builder.build(since=since) - self._cached_summaries[user_id] = { - "narrative": getattr(summary, "narrative", "") or "", - "insights": [str(i) for i in (getattr(summary, "insights", []) or [])[:8]], - "fired_at": datetime.now(timezone.utc).isoformat(), - "trigger": "manual", - "mode": mode, - } + with self._pool_lock: + self._cached_summaries[user_id] = { + "narrative": getattr(summary, "narrative", "") or "", + "insights": [str(i) for i in (getattr(summary, "insights", []) or [])[:8]], + "fired_at": datetime.now(timezone.utc).isoformat(), + "trigger": "manual", + "mode": mode, + } except Exception as e: logger.warning(f"wake-summary build failed for {user_id!r}: {e}") try: @@ -259,6 +374,10 @@ def fire_sleep_now(self, user_id: str, mode: str = "deep") -> Dict[str, Any]: ) except Exception: pass + try: + self.persist_user(user_id, engine) + except SnapshotWriteError as exc: + return {"ok": False, "error": str(exc)} return {"ok": True, **stats} @staticmethod @@ -282,19 +401,22 @@ def _build_engine(user_id: str): ) def cached_summary(self, user_id: str) -> Optional[Any]: - return self._cached_summaries.get(user_id) + with self._pool_lock: + return self._cached_summaries.get(user_id) def clear_cached_summary(self, user_id: str) -> None: - self._cached_summaries.pop(user_id, None) + with self._pool_lock: + self._cached_summaries.pop(user_id, None) # ─── Ephemeral task-context state ─────────────────────────────────── def _task_context_for(self, user_id: str) -> TaskContextState: - state = self._task_context.get(user_id) - if state is None: - state = TaskContextState() - self._task_context[user_id] = state - return state + with self._pool_lock: + state = self._task_context.get(user_id) + if state is None: + state = TaskContextState() + self._task_context[user_id] = state + return state def ingest_task_message( self, @@ -330,18 +452,43 @@ def start(self) -> None: self._sweeper.start() logger.info(f"[scm.mcp] idle sweeper started (threshold={self._idle_threshold}s)") - def stop(self) -> None: + def stop(self, timeout: float = 30.0) -> bool: + """Stop workers after draining accepted writes and flushing snapshots.""" + deadline = time.monotonic() + max(0.0, timeout) + with self._pool_lock: + self._accepting = False + users = list(self._ingest_queues) + + drained = True + for user_id in users: + remaining = max(0.0, deadline - time.monotonic()) + if not self.wait_for_pending(user_id, timeout=remaining): + drained = False + self._stop_flag.set() - # Signal each ingest worker to exit by enqueueing the sentinel. - for q in self._ingest_queues.values(): + with self._pool_lock: + queues = list(self._ingest_queues.values()) + workers = list(self._ingest_workers.values()) + engines = list(self._engines.items()) + # Signal each idle worker to exit. Accepted work is already drained. + for q in queues: try: - q.put(None) + q.put_nowait(None) except Exception: pass if self._sweeper is not None: - self._sweeper.join(timeout=5) - for t in self._ingest_workers.values(): - t.join(timeout=2) + self._sweeper.join(timeout=max(0.0, deadline - time.monotonic())) + for worker in workers: + worker.join(timeout=max(0.0, deadline - time.monotonic())) + if worker.is_alive(): + drained = False + + for user_id, engine in engines: + try: + self.persist_user(user_id, engine) + except Exception: + drained = False + return drained def _sweep_loop(self) -> None: tick = 0 @@ -375,13 +522,17 @@ def _sweep_once(self) -> int: fired = 0 now = time.time() - for user_id in list(self._last_activity.keys()): + with self._pool_lock: + user_ids = list(self._last_activity.keys()) + for user_id in user_ids: # Already-cached + unconsumed wake summary? Don't re-fire. - if user_id in self._cached_summaries: - continue + with self._pool_lock: + if user_id in self._cached_summaries: + continue + last_activity = self._last_activity.get(user_id, now) cfg = sqlite.get_user_sleep_config(user_id) mode = self._mode_from_cfg(cfg) - idle_for = max(0.0, now - self._last_activity[user_id]) + idle_for = max(0.0, now - last_activity) if mode == "off": continue @@ -448,13 +599,16 @@ def _mode_from_cfg(cfg: Dict[str, Any]) -> str: def _fire_sleep_for( self, user_id: str, idle_for: float, scheduled: bool = False, ) -> None: - with self._sleep_lock: - engine = self._engines.get(user_id) + with self.user_operation(user_id): + with self._pool_lock: + engine = self._engines.get(user_id) if engine is None: return reason = "scheduled (sleep window)" if scheduled else f"idle for {idle_for:.0f}s" logger.info( - f"[scm.mcp] firing autonomous deep-sleep for user {user_id!r} ({reason})" + "[scm.mcp] firing autonomous deep-sleep for user_key=%s (%s)", + self._state_store.user_key(user_id)[:12], + reason, ) try: engine.force_sleep("deep") @@ -470,19 +624,31 @@ def _fire_sleep_for( lookback_sec = 86400 if scheduled else (idle_for + 60) since = datetime.now(timezone.utc) - timedelta(seconds=lookback_sec) summary = engine._wake_summary_builder.build(since=since) - self._cached_summaries[user_id] = { - "narrative": getattr(summary, "narrative", "") or "", - "insights": [str(i) for i in (getattr(summary, "insights", []) or [])[:8]], - "fired_at": datetime.now(timezone.utc).isoformat(), - "idle_seconds": idle_for, - "scheduled": scheduled, - } + with self._pool_lock: + self._cached_summaries[user_id] = { + "narrative": getattr(summary, "narrative", "") or "", + "insights": [str(i) for i in (getattr(summary, "insights", []) or [])[:8]], + "fired_at": datetime.now(timezone.utc).isoformat(), + "idle_seconds": idle_for, + "scheduled": scheduled, + } logger.info( - f"[scm.mcp] wake-summary cached for {user_id!r}: " - f"{len(self._cached_summaries[user_id]['insights'])} insights" + "[scm.mcp] wake-summary cached for user_key=%s", + self._state_store.user_key(user_id)[:12], ) except Exception as e: - logger.warning(f"wake-summary build failed for {user_id!r}: {e}") + logger.warning( + "wake-summary build failed for user_key=%s: %s", + self._state_store.user_key(user_id)[:12], + type(e).__name__, + ) + try: + self.persist_user(user_id, engine) + except Exception: + logger.exception( + "snapshot persistence failed for user_key=%s", + self._state_store.user_key(user_id)[:12], + ) # ─── MCP server ─────────────────────────────────────────────────────────── @@ -518,11 +684,6 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> list: }))] user_id = arguments.get("user_id") or "default" - engine = pool.get_or_create( - user_id, - bump_activity=name in ("add_memory", "search_memory"), - ) - if name in ("sleep", "consolidate"): mode = arguments.get("mode") or "deep" result = await asyncio.get_event_loop().run_in_executor( @@ -535,7 +696,13 @@ async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> list: # stall the MCP event loop. loop = asyncio.get_event_loop() result = await loop.run_in_executor( - None, lambda: tool.handler(arguments, engine) + None, + lambda: pool.call_handler( + name, + arguments, + tool.handler, + bump_activity=name in ("add_memory", "search_memory"), + ), ) # Auto-surface a cached wake-summary on the next activity after idle. diff --git a/src/integrations/memories_api.py b/src/integrations/memories_api.py index bf39143..b4bc51a 100644 --- a/src/integrations/memories_api.py +++ b/src/integrations/memories_api.py @@ -23,12 +23,15 @@ """ from __future__ import annotations +import json import os -from typing import Any, Dict, Optional +from typing import Any, Dict, Literal, Optional -from fastapi import APIRouter, HTTPException, Query, Request +from fastapi import APIRouter, HTTPException, Path, Query, Request +from pydantic import BaseModel, ConfigDict, Field, field_validator -from .mcp_server import UserEnginePool +from .mcp_server import IngestQueueFull, UserEnginePool +from .user_state_store import SnapshotWriteError from .tools import ( TOOLS, export_all_anthropic, @@ -41,6 +44,48 @@ router = APIRouter(prefix="/v1", tags=["memories"]) +MAX_USER_ID_CHARS = 256 +MAX_MEMORY_TEXT_CHARS = 65_536 +MAX_QUERY_CHARS = 8_192 +MAX_METADATA_BYTES = 65_536 +MAX_MEMORY_ID_CHARS = 512 + + +class _StrictRequest(BaseModel): + model_config = ConfigDict( + extra="forbid", + strict=True, + str_strip_whitespace=True, + ) + + +class AddMemoryRequest(_StrictRequest): + text: str = Field(min_length=1, max_length=MAX_MEMORY_TEXT_CHARS) + user_id: str = Field(default="default", min_length=1, max_length=MAX_USER_ID_CHARS) + metadata: Dict[str, Any] = Field(default_factory=dict) + replaces_prior: bool = False + sync: bool = False + + @field_validator("metadata") + @classmethod + def metadata_is_bounded(cls, value: Dict[str, Any]) -> Dict[str, Any]: + encoded = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") + if len(encoded) > MAX_METADATA_BYTES: + raise ValueError(f"metadata exceeds {MAX_METADATA_BYTES} bytes") + return value + + +class SearchMemoryRequest(_StrictRequest): + query: str = Field(min_length=1, max_length=MAX_QUERY_CHARS) + user_id: str = Field(default="default", min_length=1, max_length=MAX_USER_ID_CHARS) + limit: int = Field(default=5, ge=1, le=50) + wait_for_pending: bool = False + + +class SleepRequest(_StrictRequest): + user_id: str = Field(default="default", min_length=1, max_length=MAX_USER_ID_CHARS) + mode: Literal["deep", "micro"] = "deep" + # Module-level singleton — one engine pool per server process. _pool: Optional[UserEnginePool] = None @@ -61,6 +106,15 @@ def get_pool() -> UserEnginePool: return _pool +def close_pool(timeout: float = 30.0) -> bool: + global _pool + pool = _pool + _pool = None + if pool is None: + return True + return pool.stop(timeout=timeout) + + def _user_id_from(payload: Dict[str, Any], header_user: Optional[str]) -> str: return (payload.get("user_id") or header_user or "default") @@ -117,9 +171,14 @@ def _invoke(tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]: text = (args.get("text") or "").strip() if not text: raise HTTPException(status_code=400, detail="'text' is required") - placeholder = pool.enqueue_ingest( - user_id, text, args.get("metadata") or {} - ) + try: + placeholder = pool.enqueue_ingest( + user_id, text, args.get("metadata") or {} + ) + except IngestQueueFull as exc: + raise HTTPException(status_code=429, detail=str(exc)) from exc + except RuntimeError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc return { "ok": True, "user_id": user_id, @@ -130,12 +189,14 @@ def _invoke(tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]: } # Sync path: drain queue first so we have a stable engine state, # then run the handler (which calls engine.chat synchronously). - pool.wait_for_pending(user_id, timeout=10.0) + if not pool.wait_for_pending(user_id, timeout=10.0): + raise HTTPException(status_code=503, detail="pending ingest drain timed out") # ── Special path: sleep goes through the pool's wake cache ─────────── if tool_name in {"sleep", "consolidate"}: # Drain pending ingests so consolidation sees the most recent state. - pool.wait_for_pending(user_id, timeout=10.0) + if not pool.wait_for_pending(user_id, timeout=10.0): + raise HTTPException(status_code=503, detail="pending ingest drain timed out") mode = args.get("mode") or "deep" result = pool.fire_sleep_now(user_id, mode=mode) result["user_id"] = user_id @@ -143,10 +204,18 @@ def _invoke(tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]: # ── Optional: search waits for pending ingests (read-your-writes) ──── if tool_name == "search_memory" and args.get("wait_for_pending"): - pool.wait_for_pending(user_id, timeout=10.0) + if not pool.wait_for_pending(user_id, timeout=10.0): + raise HTTPException(status_code=503, detail="pending ingest drain timed out") - engine = pool.get_or_create(user_id, bump_activity=bump_activity) - result = tool.handler(args, engine) + try: + result = pool.call_handler( + tool_name, + args, + tool.handler, + bump_activity=bump_activity, + ) + except SnapshotWriteError as exc: + raise HTTPException(status_code=503, detail=str(exc)) from exc # Auto-surface a cached wake-summary on activity-resumption tools. if tool_name in _USER_ACTIVITY_TOOLS: @@ -176,28 +245,26 @@ def _memory_lineage_for_user(user_id: str, memory_id: str) -> Dict[str, Any]: @router.get("/health") async def health() -> Dict[str, Any]: pool = get_pool() + snapshot = pool.health_snapshot() return { "ok": True, - "active_users": len(pool._engines), "auto_sleep": pool._auto_sleep, "idle_threshold_sec": pool._idle_threshold, + **snapshot, } @router.post("/memories") -async def add_memory(payload: Dict[str, Any], request: Request) -> Dict[str, Any]: +async def add_memory(payload: AddMemoryRequest, request: Request) -> Dict[str, Any]: """Add a memory. Body: {text, user_id?, metadata?}.""" - if "text" not in payload or not payload["text"]: - raise HTTPException(status_code=400, detail="'text' is required") - return _invoke("add_memory", _namespace_for_account(request, payload)) + args = payload.model_dump() + return _invoke("add_memory", _namespace_for_account(request, args)) @router.post("/memories/search") -async def search_memory(payload: Dict[str, Any], request: Request) -> Dict[str, Any]: +async def search_memory(payload: SearchMemoryRequest, request: Request) -> Dict[str, Any]: """Search memories by associative retrieval. Body: {query, user_id?, limit?}.""" - if "query" not in payload or not payload["query"]: - raise HTTPException(status_code=400, detail="'query' is required") - args = _namespace_for_account(request, payload) + args = _namespace_for_account(request, payload.model_dump()) result = _invoke("search_memory", args) user_id = args.get("user_id") or "default" slots = get_pool().task_context_snapshot(user_id) @@ -206,26 +273,26 @@ async def search_memory(payload: Dict[str, Any], request: Request) -> Dict[str, @router.post("/memories/consolidate") -async def consolidate(payload: Dict[str, Any], request: Request) -> Dict[str, Any]: +async def consolidate(payload: SleepRequest, request: Request) -> Dict[str, Any]: """Backward-compatible alias for /v1/memories/sleep.""" - return _invoke("consolidate", _namespace_for_account(request, payload or {})) + return _invoke("consolidate", _namespace_for_account(request, payload.model_dump())) @router.post("/memories/sleep") -async def sleep(payload: Dict[str, Any], request: Request) -> Dict[str, Any]: +async def sleep(payload: SleepRequest, request: Request) -> Dict[str, Any]: """Manually trigger a sleep cycle. Body: {user_id?, mode?}. Most callers do not need this — the auto-sleep sweeper fires it automatically when the user has been idle past the threshold. """ - return _invoke("sleep", _namespace_for_account(request, payload or {})) + return _invoke("sleep", _namespace_for_account(request, payload.model_dump())) @router.get("/wake-summary") async def wake_summary( request: Request, - user_id: str = Query("default"), - since_hours: float = Query(24.0, ge=0.5), + user_id: str = Query("default", min_length=1, max_length=MAX_USER_ID_CHARS), + since_hours: float = Query(24.0, ge=0.5, le=8760.0), ) -> Dict[str, Any]: """Return what the agent learned during recent idle time.""" payload = _namespace_for_account(request, {"user_id": user_id, "since_hours": since_hours}) @@ -233,7 +300,11 @@ async def wake_summary( @router.delete("/memories/{memory_id}") -async def forget(memory_id: str, request: Request, user_id: str = Query("default")) -> Dict[str, Any]: +async def forget( + request: Request, + memory_id: str = Path(..., min_length=1, max_length=MAX_MEMORY_ID_CHARS), + user_id: str = Query("default", min_length=1, max_length=MAX_USER_ID_CHARS), +) -> Dict[str, Any]: """Remove a specific memory by id.""" payload = _namespace_for_account(request, {"memory_id": memory_id, "user_id": user_id}) return _invoke("forget", payload) @@ -484,4 +555,4 @@ async def openapi_json() -> Dict[str, Any]: server_url = os.environ.get( "SCM_PUBLIC_URL", "http://localhost:8000" ) - return export_openapi_spec(server_url=f"{server_url}/v1/tools") + return export_openapi_spec(server_url=server_url.rstrip("/")) diff --git a/src/integrations/tools.py b/src/integrations/tools.py index 3d6734f..70af88d 100644 --- a/src/integrations/tools.py +++ b/src/integrations/tools.py @@ -31,6 +31,8 @@ from datetime import datetime, timedelta, timezone from typing import Any, Callable, Dict, List, Optional +from ..version import __version__ + # ─── Tool definitions ───────────────────────────────────────────────────── @@ -411,6 +413,8 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: _USER_ID_FIELD = { "type": "string", + "minLength": 1, + "maxLength": 256, "description": "Stable identifier for the end-user whose memory this is. Defaults to 'default' for single-user deployments. Use a per-user value (email, account ID) for multi-user systems.", "default": "default", } @@ -431,6 +435,8 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: "properties": { "text": { "type": "string", + "minLength": 1, + "maxLength": 65536, "description": "The fact or observation to remember, in natural language. Example: 'User prefers vegan food and lives in Seattle.'", }, "user_id": _USER_ID_FIELD, @@ -455,6 +461,7 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: }, }, "required": ["text"], + "additionalProperties": False, }, handler=_add_memory_handler, examples=[ @@ -478,6 +485,8 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: "properties": { "query": { "type": "string", + "minLength": 1, + "maxLength": 8192, "description": "The question or topic to search for. Free text.", }, "user_id": _USER_ID_FIELD, @@ -490,6 +499,7 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: }, }, "required": ["query"], + "additionalProperties": False, }, handler=_search_memory_handler, examples=[ @@ -520,6 +530,7 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: }, }, "required": [], + "additionalProperties": False, }, handler=_consolidate_handler, ), @@ -541,9 +552,11 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: "description": "How far back to summarise. 24 = since yesterday morning. 168 = past week.", "default": 24.0, "minimum": 0.5, + "maximum": 8760.0, }, }, "required": [], + "additionalProperties": False, }, handler=_wake_summary_handler, ), @@ -562,11 +575,14 @@ def _forget_handler(args: Dict[str, Any], engine: Any) -> Dict[str, Any]: "properties": { "memory_id": { "type": "string", + "minLength": 1, + "maxLength": 512, "description": "The id returned by a previous add_memory or search_memory call.", }, "user_id": _USER_ID_FIELD, }, "required": ["memory_id"], + "additionalProperties": False, }, handler=_forget_handler, ), @@ -612,37 +628,65 @@ def to_gemini_function(tool: ToolDef) -> Dict[str, Any]: } -def to_openapi_path(tool: ToolDef, base_path: str = "/v1/tools") -> Dict[str, Any]: +def to_openapi_path(tool: ToolDef, base_path: str = "/v1") -> Dict[str, Any]: """Render a ToolDef as an OpenAPI 3.1 path object. Used for ChatGPT Custom GPT 'Actions' which require an OpenAPI spec. """ - return { - f"{base_path}/{tool.name}": { - "post": { - "operationId": tool.name, - "summary": tool.description, - "requestBody": { - "required": True, - "content": { - "application/json": { - "schema": tool.input_schema, - }, - }, - }, - "responses": { - "200": { - "description": f"{tool.name} response", - "content": { - "application/json": { - "schema": {"type": "object"}, - }, - }, - }, + routes = { + "add_memory": ("post", f"{base_path}/memories"), + "search_memory": ("post", f"{base_path}/memories/search"), + "sleep": ("post", f"{base_path}/memories/sleep"), + "wake_summary": ("get", f"{base_path}/wake-summary"), + "forget": ("delete", f"{base_path}/memories/{{memory_id}}"), + } + method, path = routes[tool.name] + operation: Dict[str, Any] = { + "operationId": tool.name, + "summary": tool.description, + "responses": { + "200": { + "description": f"{tool.name} response", + "content": { + "application/json": {"schema": {"type": "object"}}, }, }, }, } + if method == "post": + operation["requestBody"] = { + "required": True, + "content": { + "application/json": {"schema": tool.input_schema}, + }, + } + elif tool.name == "wake_summary": + operation["parameters"] = [ + { + "name": name, + "in": "query", + "required": False, + "schema": schema, + } + for name, schema in tool.input_schema["properties"].items() + ] + else: + properties = tool.input_schema["properties"] + operation["parameters"] = [ + { + "name": "memory_id", + "in": "path", + "required": True, + "schema": properties["memory_id"], + }, + { + "name": "user_id", + "in": "query", + "required": False, + "schema": properties["user_id"], + }, + ] + return {path: {method: operation}} def export_all_openai() -> List[Dict[str, Any]]: @@ -672,7 +716,7 @@ def export_openapi_spec(server_url: str = "https://api.scm.example.com") -> Dict "(attention, encoding, retrieval) plus sleep phase " "(consolidation, schema extraction, knowledge-gap filling)." ), - "version": "0.9.2", + "version": __version__, }, "servers": [{"url": server_url}], "paths": paths, diff --git a/src/integrations/user_state_store.py b/src/integrations/user_state_store.py new file mode 100644 index 0000000..b18f7f8 --- /dev/null +++ b/src/integrations/user_state_store.py @@ -0,0 +1,200 @@ +"""Crash-safe per-user snapshots for the public REST and MCP runtimes.""" +from __future__ import annotations + +import hashlib +import hmac +import json +import os +import threading +import uuid +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Dict, Optional + + +SNAPSHOT_FORMAT_VERSION = 1 +DEFAULT_MAX_SNAPSHOT_BYTES = 64 * 1024 * 1024 + + +class SnapshotError(RuntimeError): + """Base class for snapshot persistence failures.""" + + +class SnapshotWriteError(SnapshotError): + """Raised when an atomic snapshot cannot be committed.""" + + +@dataclass(frozen=True) +class SnapshotLoadResult: + loaded: bool + status: str + stats: Optional[Dict[str, int]] = None + quarantine_path: Optional[Path] = None + + +def _json_default(value: Any) -> str: + try: + return value.isoformat() + except Exception: + return str(value) + + +class UserStateStore: + """Persist isolated engine exports without exposing user IDs as paths. + + The legacy SQLite layer is process-global, so the multi-user product + runtime keeps engines in sandbox mode. This store adds durability around + those isolated engines using checksummed, atomic JSON snapshots. + """ + + def __init__( + self, + root: Optional[Path] = None, + enabled: bool = True, + max_snapshot_bytes: int = DEFAULT_MAX_SNAPSHOT_BYTES, + ) -> None: + data_dir = Path( + os.environ.get("SCM_DATA_DIR", str(Path.home() / ".scm")) + ).expanduser() + self.root = Path(root) if root is not None else data_dir / "users" + self.enabled = bool(enabled) + self.max_snapshot_bytes = int(max_snapshot_bytes) + self._locks_guard = threading.Lock() + self._locks: Dict[str, threading.RLock] = {} + if self.enabled: + self.root.mkdir(parents=True, exist_ok=True) + + @staticmethod + def user_key(user_id: str) -> str: + normalized = str(user_id or "default").encode("utf-8") + return hashlib.sha256(normalized).hexdigest() + + def snapshot_path(self, user_id: str) -> Path: + return self.root / f"{self.user_key(user_id)}.json" + + def _lock_for(self, user_id: str) -> threading.RLock: + key = self.user_key(user_id) + with self._locks_guard: + return self._locks.setdefault(key, threading.RLock()) + + @staticmethod + def _memory_bytes(memory: Dict[str, Any]) -> bytes: + return json.dumps( + memory, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=_json_default, + ).encode("utf-8") + + def save(self, user_id: str, engine: Any) -> Optional[Path]: + if not self.enabled: + return None + + memory = engine.export_memory() + if not isinstance(memory, dict): + raise SnapshotWriteError("engine export must be a JSON object") + + memory_bytes = self._memory_bytes(memory) + envelope = { + "format_version": SNAPSHOT_FORMAT_VERSION, + "user_key": self.user_key(user_id), + "saved_at": datetime.now(timezone.utc).isoformat(), + "checksum_sha256": hashlib.sha256(memory_bytes).hexdigest(), + "memory": memory, + } + payload = json.dumps( + envelope, + sort_keys=True, + separators=(",", ":"), + ensure_ascii=False, + default=_json_default, + ).encode("utf-8") + if len(payload) > self.max_snapshot_bytes: + raise SnapshotWriteError( + f"snapshot exceeds {self.max_snapshot_bytes} byte safety limit" + ) + + path = self.snapshot_path(user_id) + temp_path = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp") + with self._lock_for(user_id): + try: + self.root.mkdir(parents=True, exist_ok=True) + with temp_path.open("xb") as handle: + handle.write(payload) + handle.flush() + os.fsync(handle.fileno()) + os.replace(temp_path, path) + self._fsync_directory() + except Exception as exc: + try: + temp_path.unlink(missing_ok=True) + except Exception: + pass + raise SnapshotWriteError( + f"failed to commit snapshot: {type(exc).__name__}" + ) from exc + return path + + def load(self, user_id: str, engine: Any) -> SnapshotLoadResult: + if not self.enabled: + return SnapshotLoadResult(loaded=False, status="disabled") + + path = self.snapshot_path(user_id) + with self._lock_for(user_id): + if not path.exists(): + return SnapshotLoadResult(loaded=False, status="missing") + try: + size = path.stat().st_size + if size <= 0 or size > self.max_snapshot_bytes: + raise SnapshotError("snapshot size is invalid") + envelope = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(envelope, dict): + raise SnapshotError("snapshot envelope is invalid") + if envelope.get("format_version") != SNAPSHOT_FORMAT_VERSION: + raise SnapshotError("snapshot format is unsupported") + if envelope.get("user_key") != self.user_key(user_id): + raise SnapshotError("snapshot identity does not match") + memory = envelope.get("memory") + if not isinstance(memory, dict): + raise SnapshotError("snapshot memory is invalid") + expected = envelope.get("checksum_sha256") + actual = hashlib.sha256(self._memory_bytes(memory)).hexdigest() + if not expected or not hmac.compare_digest(str(expected), actual): + raise SnapshotError("snapshot checksum does not match") + stats = engine.import_memory(memory, replace_existing=True) + if not isinstance(stats, dict): + stats = {} + return SnapshotLoadResult( + loaded=True, + status="loaded", + stats={str(k): int(v) for k, v in stats.items()}, + ) + except Exception: + quarantine = self._quarantine(path) + return SnapshotLoadResult( + loaded=False, + status="corrupt", + quarantine_path=quarantine, + ) + + def _quarantine(self, path: Path) -> Optional[Path]: + stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ") + target = path.with_name(f"{path.name}.corrupt.{stamp}") + try: + os.replace(path, target) + self._fsync_directory() + return target + except Exception: + return None + + def _fsync_directory(self) -> None: + try: + descriptor = os.open(self.root, os.O_RDONLY) + except Exception: + return + try: + os.fsync(descriptor) + finally: + os.close(descriptor) diff --git a/src/sleep/forgetting_dynamics.py b/src/sleep/forgetting_dynamics.py index c7ddfe4..f59da1d 100644 --- a/src/sleep/forgetting_dynamics.py +++ b/src/sleep/forgetting_dynamics.py @@ -4,7 +4,7 @@ from __future__ import annotations from collections import Counter -from datetime import datetime, timedelta +from datetime import datetime, timedelta, timezone import math from typing import Dict, List, Optional, Tuple @@ -94,7 +94,12 @@ def apply( """ Update retention, decay, and state for a batch of concepts. """ - now = ensure_utc(now) or utc_now() + if now is None: + now = utc_now() + elif now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + elif now.utcoffset() != timedelta(0): + now = now.astimezone(timezone.utc) conflict_pairs = conflict_pairs or [] conflict_counts = Counter() for left, right in conflict_pairs: @@ -190,8 +195,14 @@ def compute_retention_score( R = p1*Grasp + p2*Salience + p3*Rehearsal + p4*AssociationDensity + p5*Recency - p6*Interference """ - now = ensure_utc(now) or utc_now() - conflict_counts = conflict_counts or Counter() + if now is None: + now = utc_now() + elif now.tzinfo is None: + now = now.replace(tzinfo=timezone.utc) + elif now.utcoffset() != timedelta(0): + now = now.astimezone(timezone.utc) + if conflict_counts is None: + conflict_counts = Counter() grasp = self._clamp(self._concept_value(concept, "grasp_score", fallback_importance=True)) salience = self._clamp(self._concept_value(concept, "salience_score", fallback_importance=True)) @@ -256,21 +267,121 @@ def get_forgetting_stats(self, concepts: List[Concept]) -> Dict: "current_threshold": self.suppress_threshold, } - retentions = [] - decays = [] + # This path backs health/readiness probes and can run frequently. Keep + # the exact retention formula, but aggregate in one pass instead of + # allocating per-concept Counters/lists and repeatedly normalizing the + # same clock value. + now = utc_now() + retention_total = 0.0 + decay_total = 0.0 + retentions_below_threshold = 0 + threshold = self.compute_forgetting_threshold(concepts) + + grasp_weight = self.grasp_weight + salience_weight = self.salience_weight + rehearsal_weight = self.rehearsal_weight + association_weight = self.association_weight + recency_weight = self.recency_weight + interference_weight = self.interference_weight + base_decay = self.base_decay + for concept in concepts: - retention, _ = self.compute_retention_score(concept) - decays.append(self.compute_decay_lambda(concept, retention)) - retentions.append(retention) + importance = getattr(concept, "importance", None) + importance_overall = float(importance.overall) if importance is not None else 0.5 + rehearsal_count = getattr(concept, "rehearsal_count", 0) + activation_count = getattr(concept, "activation_count", 0) + association = max( + 0.0, + min(1.0, float(getattr(concept, "association_density", 0.0))), + ) + generic_trace = ( + rehearsal_count == 0 + and activation_count == 0 + and association <= 0.05 + ) - threshold = self.compute_forgetting_threshold(concepts) - forgettable = sum(1 for retention in retentions if retention < threshold) + raw_grasp = getattr(concept, "grasp_score", None) + if raw_grasp is None or ( + abs(float(raw_grasp) - 0.5) < 1e-6 and generic_trace + ): + raw_grasp = importance_overall + grasp = max(0.0, min(1.0, float(raw_grasp))) + + raw_salience = getattr(concept, "salience_score", None) + if raw_salience is None or ( + abs(float(raw_salience) - 0.5) < 1e-6 and generic_trace + ): + raw_salience = importance_overall + salience = max(0.0, min(1.0, float(raw_salience))) + rehearsal = max(0.0, min(1.0, rehearsal_count / 8.0)) + + last_accessed = getattr(concept, "last_accessed", None) + if last_accessed: + if not isinstance(last_accessed, datetime): + last_accessed = ensure_utc(last_accessed) + elif last_accessed.tzinfo is None: + last_accessed = last_accessed.replace(tzinfo=timezone.utc) + elif last_accessed.utcoffset() != timedelta(0): + last_accessed = last_accessed.astimezone(timezone.utc) + if not isinstance(last_accessed, datetime): + recency = 0.5 + else: + age_hours = max( + 0.0, + (now - last_accessed).total_seconds() / 3600.0, + ) + if age_hours <= 1.0: + recency = 1.0 + elif age_hours >= 72.0: + recency = 0.1 + else: + recency = max(0.0, min(1.0, 1.0 - (age_hours - 1.0) / 71.0)) + + state = self._state_value(concept) + interference = 0.0 + if getattr(concept, "version_parent", None): + interference += 0.20 if not getattr(concept, "is_current_version", True) else 0.05 + if state == MemoryState.SUPPRESSED.value: + interference += 0.10 + if state == MemoryState.ARCHIVED.value: + interference += 0.20 + interference = max(0.0, min(1.0, interference)) + + retention = max( + 0.0, + min( + 1.0, + grasp_weight * grasp + + salience_weight * salience + + rehearsal_weight * rehearsal + + association_weight * association + + recency_weight * recency + - interference_weight * interference, + ), + ) + retention_total += retention + if retention < threshold: + retentions_below_threshold += 1 + + is_current = getattr(concept, "is_current_version", True) + version_penalty = 0.15 if not is_current else 0.0 + state_penalty = 0.10 if state == MemoryState.SUPPRESSED.value else 0.0 + decay = base_decay * ( + 1.0 + + (1.0 - retention) * 1.5 + + interference * 0.8 + + version_penalty + + state_penalty + ) + decay_total += max(0.005, round(decay, 4)) + + total = len(concepts) return { - "total_concepts": len(concepts), - "forgettable": forgettable, - "preserve": len(concepts) - forgettable, - "avg_retention": round(sum(retentions) / len(retentions), 4), - "avg_decay": round(sum(decays) / len(decays), 4), + "total_concepts": total, + "forgettable": retentions_below_threshold, + "preserve": total - retentions_below_threshold, + "avg_retention": round(retention_total / total, 4), + "avg_decay": round(decay_total / total, 4), "current_threshold": threshold, } @@ -368,7 +479,12 @@ def _age_hours(self, concept: Concept, now: datetime) -> float: last_accessed = getattr(concept, "last_accessed", None) if not last_accessed: return 0.0 - last_accessed = ensure_utc(last_accessed) + if not isinstance(last_accessed, datetime): + last_accessed = ensure_utc(last_accessed) + elif last_accessed.tzinfo is None: + last_accessed = last_accessed.replace(tzinfo=timezone.utc) + elif last_accessed.utcoffset() != timedelta(0): + last_accessed = last_accessed.astimezone(timezone.utc) if not isinstance(last_accessed, datetime): return 0.0 return max(0.0, (now - last_accessed).total_seconds() / 3600.0) diff --git a/src/sleep/sleep_cycle.py b/src/sleep/sleep_cycle.py index b350a12..b54a2d2 100644 --- a/src/sleep/sleep_cycle.py +++ b/src/sleep/sleep_cycle.py @@ -492,7 +492,39 @@ def get_sleep_readiness(self, concepts: List[Concept], relations: List[Relation] forgetting_stats = self.forgetting.get_forgetting_stats(concepts) time_since = self.trigger.time_since_last_sleep() - mode, reason, _ = self.select_sleep_mode(concepts, relations) + deep_idle = self.trigger.time_since_last_deep_sleep() + entropy = trigger_stats["current_entropy"] + conflict = trigger_stats["conflict_density"] + pressure = max(entropy, conflict) + + mode: Optional[str] = None + reason = "Sleep not needed" + if concepts and pressure >= self.trigger.deep_pressure_threshold: + mode = "deep" + reason = ( + f"Deep pressure high: {pressure:.3f} >= " + f"{self.trigger.deep_pressure_threshold}" + ) + elif ( + concepts + and deep_idle is not None + and deep_idle >= self.trigger.deep_min_idle_seconds + ): + mode = "deep" + reason = ( + f"Deep idle window reached: {deep_idle:.1f}s >= " + f"{self.trigger.deep_min_idle_seconds}s" + ) + elif concepts and self.micro_sleep_enabled: + if entropy >= self.trigger.micro_entropy_threshold: + mode = "micro" + reason = ( + f"Micro entropy spike: {entropy:.3f} >= " + f"{self.trigger.micro_entropy_threshold}" + ) + elif conflict >= (self.trigger.conflict_threshold * 0.85): + mode = "micro" + reason = f"Micro conflict pressure: {conflict:.3f}" return { 'entropy': trigger_stats['current_entropy'], diff --git a/src/version.py b/src/version.py new file mode 100644 index 0000000..659c421 --- /dev/null +++ b/src/version.py @@ -0,0 +1,3 @@ +"""Single runtime version constant; packaging parity is enforced in tests.""" + +__version__ = "0.9.2" diff --git a/tests/agent_with_tools/test_supervisor_team.py b/tests/agent_with_tools/test_supervisor_team.py index 73ccffa..5de7bb0 100644 --- a/tests/agent_with_tools/test_supervisor_team.py +++ b/tests/agent_with_tools/test_supervisor_team.py @@ -26,7 +26,13 @@ import uuid from typing import Annotated +import pytest from dotenv import load_dotenv + +pytest.importorskip("langchain_core", reason="LangChain integration extra is not installed") +pytest.importorskip("langchain_openai", reason="LangChain OpenAI integration is not installed") +pytest.importorskip("langgraph", reason="LangGraph integration is not installed") + load_dotenv() # Add repo root to sys.path so `src.*` imports resolve. diff --git a/tests/production/__init__.py b/tests/production/__init__.py new file mode 100644 index 0000000..33888ac --- /dev/null +++ b/tests/production/__init__.py @@ -0,0 +1 @@ +"""Deterministic product release qualification tests.""" diff --git a/tests/production/conftest.py b/tests/production/conftest.py new file mode 100644 index 0000000..418a37d --- /dev/null +++ b/tests/production/conftest.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +import importlib + +import pytest +from fastapi.testclient import TestClient + + +@pytest.fixture(autouse=True) +def isolated_product_runtime(tmp_path, monkeypatch): + data_dir = tmp_path / "scm-data" + monkeypatch.setenv("SCM_DATA_DIR", str(data_dir)) + monkeypatch.setenv("SCM_API_PERSISTENCE", "1") + monkeypatch.setenv("SCM_AUTO_SLEEP_DISABLE", "1") + monkeypatch.setenv("IDLE_LEARNER_ENABLED", "false") + monkeypatch.setenv("LLM_PROVIDER", "") + monkeypatch.setenv("SCM_EMBEDDING_BACKEND", "hash") + monkeypatch.setenv("SCM_CLOUD_AUTH", "0") + for key in ( + "OPENAI_API_KEY", + "DEEPSEEK_API_KEY", + "ANTHROPIC_API_KEY", + "NPM_TOKEN", + "TWINE_PASSWORD", + ): + monkeypatch.delenv(key, raising=False) + + from src.core import sqlite_db + from src.integrations import memories_api + + previous_db_path = sqlite_db.DB_PATH + memories_api.close_pool(timeout=5.0) + sqlite_db.set_db_path(data_dir / "sleepai.db") + + import src.cloud.accounts as accounts + + accounts._SCHEMA_ENSURED = False + yield + + memories_api.close_pool(timeout=10.0) + sqlite_db.set_db_path(previous_db_path) + importlib.reload(accounts) + + +@pytest.fixture +def product_client(): + from src.api.main import app + + with TestClient(app) as client: + yield client diff --git a/tests/production/test_api_abuse.py b/tests/production/test_api_abuse.py new file mode 100644 index 0000000..9825e84 --- /dev/null +++ b/tests/production/test_api_abuse.py @@ -0,0 +1,104 @@ +from __future__ import annotations + +import json + +import pytest +from hypothesis import HealthCheck, given, settings, strategies as st + + +pytestmark = [pytest.mark.production, pytest.mark.security] + + +@pytest.mark.parametrize( + "payload", + [ + {}, + {"text": None}, + {"text": 42}, + {"text": []}, + {"text": ""}, + {"text": " "}, + {"text": "ok", "unknown": True}, + {"text": "x" * 65_537}, + {"text": "ok", "user_id": "u" * 257}, + {"text": "ok", "sync": "true"}, + ], +) +def test_add_memory_rejects_malformed_payloads_without_5xx(product_client, payload): + response = product_client.post("/v1/memories", json=payload) + assert response.status_code == 422 + + +@pytest.mark.parametrize( + "payload", + [ + {}, + {"query": None}, + {"query": 7}, + {"query": ""}, + {"query": "x", "limit": 0}, + {"query": "x", "limit": 51}, + {"query": "x", "limit": "5"}, + {"query": "x", "wait_for_pending": "yes"}, + ], +) +def test_search_rejects_malformed_payloads_without_5xx(product_client, payload): + response = product_client.post("/v1/memories/search", json=payload) + assert response.status_code == 422 + + +def test_sleep_and_wake_ranges_are_strict(product_client): + assert product_client.post( + "/v1/memories/sleep", json={"mode": "ultra"} + ).status_code == 422 + assert product_client.get( + "/v1/wake-summary", params={"since_hours": 100_000} + ).status_code == 422 + + +def test_request_body_limit_fails_closed(product_client): + body = json.dumps({"text": "x" * 1_100_000}).encode("utf-8") + response = product_client.post( + "/v1/memories", + content=body, + headers={"Content-Type": "application/json"}, + ) + assert response.status_code == 413 + + +def test_unicode_memory_round_trip(product_client): + text = "姓名是美咲. Cafe preference: creme brulee. Arabic: مرحبا." + added = product_client.post( + "/v1/memories", + json={"user_id": "unicode-user", "text": text, "sync": True}, + ) + assert added.status_code == 200 + found = product_client.post( + "/v1/memories/search", + json={"user_id": "unicode-user", "query": "creme brulee", "wait_for_pending": True}, + ) + assert found.status_code == 200 + assert found.json()["ok"] is True + + +json_scalars = st.one_of(st.none(), st.booleans(), st.integers(), st.floats(allow_nan=False), st.text()) +json_values = st.recursive( + json_scalars, + lambda children: st.one_of( + st.lists(children, max_size=5), + st.dictionaries(st.text(max_size=20), children, max_size=5), + ), + max_leaves=15, +) + + +@given(payload=json_values) +@settings( + max_examples=40, + deadline=None, + derandomize=True, + suppress_health_check=[HealthCheck.function_scoped_fixture], +) +def test_arbitrary_json_never_turns_client_error_into_server_crash(product_client, payload): + response = product_client.post("/v1/memories", json=payload) + assert response.status_code < 500 diff --git a/tests/production/test_concurrency.py b/tests/production/test_concurrency.py new file mode 100644 index 0000000..2b1eb81 --- /dev/null +++ b/tests/production/test_concurrency.py @@ -0,0 +1,152 @@ +from __future__ import annotations + +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from src.integrations.mcp_server import IngestQueueFull, UserEnginePool +from src.integrations.user_state_store import UserStateStore + + +pytestmark = [pytest.mark.production, pytest.mark.load] + + +class CountingEngine: + def __init__(self, delay: float = 0.0, gate: threading.Event | None = None): + self.delay = delay + self.gate = gate + self.calls: list[str] = [] + self._lock = threading.Lock() + + def chat(self, text: str): + if self.gate is not None: + self.gate.wait(timeout=5.0) + if self.delay: + time.sleep(self.delay) + with self._lock: + self.calls.append(text) + return "", {"concepts_added": 1} + + def export_memory(self): + with self._lock: + return {"calls": list(self.calls)} + + def import_memory(self, payload, replace_existing=True): + with self._lock: + self.calls = list(payload.get("calls", [])) + return {"concepts_imported": len(self.calls)} + + +def _pool(tmp_path, **kwargs) -> UserEnginePool: + store = UserStateStore(root=tmp_path / "users", enabled=False) + return UserEnginePool(auto_sleep=False, state_store=store, **kwargs) + + +def test_engine_is_constructed_exactly_once_under_creation_race(tmp_path, monkeypatch): + pool = _pool(tmp_path) + builds = 0 + build_lock = threading.Lock() + + def build(_user_id): + nonlocal builds + time.sleep(0.01) + with build_lock: + builds += 1 + return CountingEngine() + + monkeypatch.setattr(pool, "_build_engine", build) + with ThreadPoolExecutor(max_workers=32) as executor: + engines = list(executor.map(lambda _: pool.get_or_create("same-user"), range(64))) + + assert builds == 1 + assert len({id(engine) for engine in engines}) == 1 + assert pool.stop(timeout=5.0) is True + + +def test_same_user_operations_are_serialized(tmp_path, monkeypatch): + pool = _pool(tmp_path) + monkeypatch.setattr(pool, "_build_engine", lambda _uid: CountingEngine()) + active = 0 + peak = 0 + guard = threading.Lock() + + def handler(_args, _engine): + nonlocal active, peak + with guard: + active += 1 + peak = max(peak, active) + time.sleep(0.005) + with guard: + active -= 1 + return {"ok": True} + + def call(_index): + return pool.call_handler( + "search_memory", {"user_id": "one"}, handler, bump_activity=True + ) + + with ThreadPoolExecutor(max_workers=16) as executor: + list(executor.map(call, range(64))) + assert peak == 1 + pool.stop(timeout=5.0) + + +def test_different_users_execute_in_parallel(tmp_path, monkeypatch): + pool = _pool(tmp_path) + monkeypatch.setattr(pool, "_build_engine", lambda _uid: CountingEngine()) + rendezvous = threading.Barrier(2) + + def handler(_args, _engine): + rendezvous.wait(timeout=2.0) + return {"ok": True} + + with ThreadPoolExecutor(max_workers=2) as executor: + futures = [ + executor.submit( + pool.call_handler, + "search_memory", + {"user_id": user_id}, + handler, + bump_activity=True, + ) + for user_id in ("alice", "bob") + ] + assert all(future.result(timeout=3.0)["ok"] for future in futures) + pool.stop(timeout=5.0) + + +def test_shutdown_drains_every_accepted_async_write(tmp_path, monkeypatch): + pool = _pool(tmp_path, max_pending_per_user=200) + engine = CountingEngine(delay=0.001) + monkeypatch.setattr(pool, "_build_engine", lambda _uid: engine) + + accepted = 100 + for index in range(accepted): + pool.enqueue_ingest("drain-user", f"memory-{index}") + + assert pool.stop(timeout=10.0) is True + assert pool.pending_count("drain-user") == 0 + assert engine.calls == [f"memory-{index}" for index in range(accepted)] + + +def test_async_queue_is_bounded_and_applies_backpressure(tmp_path, monkeypatch): + gate = threading.Event() + pool = _pool(tmp_path, max_pending_per_user=2) + monkeypatch.setattr(pool, "_build_engine", lambda _uid: CountingEngine(gate=gate)) + + pool.enqueue_ingest("bounded", "first") + pool.enqueue_ingest("bounded", "second") + with pytest.raises(IngestQueueFull): + pool.enqueue_ingest("bounded", "third") + + gate.set() + assert pool.stop(timeout=5.0) is True + + +def test_pool_rejects_writes_after_shutdown(tmp_path): + pool = _pool(tmp_path) + assert pool.stop(timeout=1.0) is True + with pytest.raises(RuntimeError, match="shutting down"): + pool.enqueue_ingest("late", "must not be accepted") diff --git a/tests/production/test_contract.py b/tests/production/test_contract.py new file mode 100644 index 0000000..27603c0 --- /dev/null +++ b/tests/production/test_contract.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +try: + import tomllib +except ModuleNotFoundError: # Python 3.10 + import tomli as tomllib +from pathlib import Path + +import jsonschema +import pytest + +from scm import SCMClient, SCMEngine, __version__ +from src.api.main import app +from src.integrations import memories_api +from src.integrations.tools import TOOLS, export_openapi_spec + + +pytestmark = pytest.mark.production +ROOT = Path(__file__).resolve().parents[2] +CANONICAL_TOOLS = ["add_memory", "search_memory", "sleep", "wake_summary", "forget"] + + +def test_version_is_identical_across_every_public_surface(): + pyproject = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8")) + package_json = json.loads((ROOT / "sdk/js/package.json").read_text(encoding="utf-8")) + openapi = export_openapi_spec() + + assert pyproject["project"]["version"] == __version__ + assert package_json["version"] == __version__ + assert app.version == __version__ + assert openapi["info"]["version"] == __version__ + + +def test_public_python_api_does_not_require_src_imports(): + assert SCMEngine.__module__.startswith("scm.") + assert SCMClient.__name__ == "SCMClient" + + +def test_tool_schemas_are_closed_valid_and_examples_conform(): + assert [tool.name for tool in TOOLS] == CANONICAL_TOOLS + for tool in TOOLS: + jsonschema.Draft202012Validator.check_schema(tool.input_schema) + assert tool.input_schema["additionalProperties"] is False + validator = jsonschema.Draft202012Validator(tool.input_schema) + for example in tool.examples: + validator.validate(example) + + +def test_exported_openapi_points_only_to_real_rest_operations(): + spec = export_openapi_spec(server_url="http://127.0.0.1:8000") + operations = { + operation["operationId"]: (method, path) + for path, methods in spec["paths"].items() + for method, operation in methods.items() + } + assert operations == { + "add_memory": ("post", "/v1/memories"), + "search_memory": ("post", "/v1/memories/search"), + "sleep": ("post", "/v1/memories/sleep"), + "wake_summary": ("get", "/v1/wake-summary"), + "forget": ("delete", "/v1/memories/{memory_id}"), + } + app_operations = { + (method.lower(), path) + for path, methods in app.openapi()["paths"].items() + for method in methods + if method.lower() in {"get", "post", "put", "patch", "delete"} + } + for method, path in operations.values(): + assert (method, path) in app_operations + + +def test_rest_lifecycle_survives_engine_pool_restart(product_client): + user_id = "restart-proof-user" + sentinel = "My recovery phrase is cobalt observatory 7319." + added = product_client.post( + "/v1/memories", + json={"user_id": user_id, "text": sentinel, "sync": True}, + ) + assert added.status_code == 200, added.text + assert added.json()["ok"] is True + + assert memories_api.close_pool(timeout=10.0) is True + + found = product_client.post( + "/v1/memories/search", + json={ + "user_id": user_id, + "query": "What is my recovery phrase?", + "wait_for_pending": True, + }, + ) + assert found.status_code == 200, found.text + payload = found.json() + rendered = json.dumps(payload, ensure_ascii=False).lower() + assert "cobalt observatory 7319" in rendered + + +def test_forget_is_durable_across_restart(product_client): + user_id = "durable-forget-user" + added = product_client.post( + "/v1/memories", + json={"user_id": user_id, "text": "My temporary code is AX-991.", "sync": True}, + ).json() + memory_id = added["memory_id"] + deleted = product_client.delete( + f"/v1/memories/{memory_id}", params={"user_id": user_id} + ) + assert deleted.status_code == 200 + assert deleted.json()["ok"] is True + + memories_api.close_pool(timeout=10.0) + found = product_client.post( + "/v1/memories/search", + json={"user_id": user_id, "query": "AX-991", "wait_for_pending": True}, + ) + assert found.status_code == 200 + assert "ax-991" not in json.dumps(found.json().get("memories", [])).lower() diff --git a/tests/production/test_performance.py b/tests/production/test_performance.py new file mode 100644 index 0000000..118b767 --- /dev/null +++ b/tests/production/test_performance.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import statistics +import time +from collections import Counter + +import pytest + +from src.core.models import Concept, ConceptType, ImportanceVector, MemoryState +from src.core.time_utils import utc_now +from src.sleep.forgetting_dynamics import ForgettingDynamics +from src.integrations.mcp_server import UserEnginePool +from src.integrations.user_state_store import UserStateStore + +from .test_concurrency import CountingEngine + + +pytestmark = [pytest.mark.production, pytest.mark.load] + + +def _percentile(values: list[float], percentile: float) -> float: + ordered = sorted(values) + index = min(len(ordered) - 1, int((len(ordered) - 1) * percentile)) + return ordered[index] + + +def test_async_ingest_acceptance_p99_stays_interactive(tmp_path, monkeypatch): + pool = UserEnginePool( + auto_sleep=False, + state_store=UserStateStore(root=tmp_path / "disabled", enabled=False), + max_pending_per_user=500, + ) + monkeypatch.setattr(pool, "_build_engine", lambda _uid: CountingEngine(delay=0.002)) + + latencies_ms = [] + for index in range(200): + started = time.perf_counter() + pool.enqueue_ingest("latency-user", f"memory-{index}") + latencies_ms.append((time.perf_counter() - started) * 1000.0) + + assert pool.stop(timeout=15.0) is True + p99 = _percentile(latencies_ms, 0.99) + assert p99 < 50.0, {"p50_ms": statistics.median(latencies_ms), "p99_ms": p99} + + +def test_health_probe_p95_budget(product_client): + latencies_ms = [] + for _ in range(200): + started = time.perf_counter() + response = product_client.get("/v1/health") + latencies_ms.append((time.perf_counter() - started) * 1000.0) + assert response.status_code == 200 + assert _percentile(latencies_ms, 0.95) < 250.0 + + +def test_optimized_forgetting_diagnostics_match_canonical_formula(): + dynamics = ForgettingDynamics() + concepts = [ + Concept( + type=ConceptType.FACT, + description=f"diagnostic-{index}", + importance=ImportanceVector( + novelty=0.2 + index * 0.1, + task_relevance=0.3 + index * 0.1, + repetition=0.1 * index, + ), + state=state, + version_parent="prior" if index == 2 else None, + is_current_version=index != 2, + ) + for index, state in enumerate( + [MemoryState.ACTIVE, MemoryState.SUPPRESSED, MemoryState.ARCHIVED] + ) + ] + now = utc_now() + canonical = [ + dynamics.compute_retention_score( + concept, + conflict_counts=Counter(), + now=now, + ) + for concept in concepts + ] + threshold = dynamics.compute_forgetting_threshold(concepts) + expected_decays = [ + dynamics.compute_decay_lambda(concept, retention, interference) + for concept, (retention, interference) in zip(concepts, canonical) + ] + + stats = dynamics.get_forgetting_stats(concepts) + assert stats["forgettable"] == sum( + 1 for retention, _ in canonical if retention < threshold + ) + assert stats["avg_retention"] == round( + sum(retention for retention, _ in canonical) / len(canonical), 4 + ) + assert stats["avg_decay"] == round(sum(expected_decays) / len(expected_decays), 4) diff --git a/tests/production/test_recovery.py b/tests/production/test_recovery.py new file mode 100644 index 0000000..88e10ea --- /dev/null +++ b/tests/production/test_recovery.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import json +import os + +import pytest + +from src.integrations.user_state_store import SnapshotWriteError, UserStateStore + + +pytestmark = [pytest.mark.production, pytest.mark.recovery] + + +class SnapshotEngine: + def __init__(self, memory=None): + self.memory = memory or {"concepts": [], "relations": []} + + def export_memory(self): + return self.memory + + def import_memory(self, payload, replace_existing=True): + self.memory = payload + return {"concepts_imported": len(payload.get("concepts", []))} + + +def test_snapshot_round_trip_is_checksummed_and_versioned(tmp_path): + store = UserStateStore(root=tmp_path / "users") + source = SnapshotEngine({"concepts": [{"id": "c1", "description": "critical"}]}) + path = store.save("alice", source) + envelope = json.loads(path.read_text(encoding="utf-8")) + + assert envelope["format_version"] == 1 + assert len(envelope["checksum_sha256"]) == 64 + assert envelope["user_key"] == store.user_key("alice") + + target = SnapshotEngine() + result = store.load("alice", target) + assert result.loaded is True + assert result.status == "loaded" + assert target.memory == source.memory + + +def test_corrupt_snapshot_is_quarantined_instead_of_crashing_startup(tmp_path): + store = UserStateStore(root=tmp_path / "users") + path = store.snapshot_path("alice") + path.write_text('{"format_version":1,"broken":', encoding="utf-8") + + result = store.load("alice", SnapshotEngine()) + assert result.loaded is False + assert result.status == "corrupt" + assert result.quarantine_path is not None + assert result.quarantine_path.exists() + assert not path.exists() + + +def test_failed_atomic_replace_preserves_last_known_good_snapshot(tmp_path, monkeypatch): + store = UserStateStore(root=tmp_path / "users") + path = store.save("alice", SnapshotEngine({"concepts": [{"id": "good"}]})) + original = path.read_bytes() + + def fail_replace(_source, _target): + raise OSError("simulated disk failure") + + monkeypatch.setattr(os, "replace", fail_replace) + with pytest.raises(SnapshotWriteError): + store.save("alice", SnapshotEngine({"concepts": [{"id": "new"}]})) + + assert path.read_bytes() == original + assert not list(path.parent.glob("*.tmp")) + + +def test_hostile_user_id_cannot_escape_snapshot_root(tmp_path): + root = tmp_path / "users" + store = UserStateStore(root=root) + hostile = "../../outside/../absolute\\windows\x00name" + path = store.save(hostile, SnapshotEngine({"concepts": []})) + + assert path.parent.resolve() == root.resolve() + assert path.name == f"{store.user_key(hostile)}.json" + assert hostile not in str(path) diff --git a/tests/production/test_security.py b/tests/production/test_security.py new file mode 100644 index 0000000..eef6af3 --- /dev/null +++ b/tests/production/test_security.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import os +import re +import subprocess +import sys +from pathlib import Path + +import pytest +from cryptography.exceptions import InvalidTag + +from src.cloud import accounts + + +pytestmark = [pytest.mark.production, pytest.mark.security] +ROOT = Path(__file__).resolve().parents[2] + + +SECRET_PATTERNS = { + "OpenAI token": re.compile(rb"sk-(?:proj-)?[A-Za-z0-9_-]{24,}"), + "PyPI token": re.compile(rb"pypi-[A-Za-z0-9_-]{40,}"), + "npm token": re.compile(rb"npm_[A-Za-z0-9]{30,}"), + "GitHub token": re.compile(rb"gh[pousr]_[A-Za-z0-9]{30,}"), + "private key": re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"), +} + + +def test_no_live_secret_patterns_in_tracked_files(): + tracked = subprocess.check_output( + ["git", "ls-files", "-z"], cwd=ROOT + ).split(b"\0") + findings = [] + for raw_name in tracked: + if not raw_name: + continue + path = ROOT / os.fsdecode(raw_name) + try: + data = path.read_bytes() + except (OSError, IsADirectoryError): + continue + for label, pattern in SECRET_PATTERNS.items(): + match = pattern.search(data) + if match: + line = data.count(b"\n", 0, match.start()) + 1 + findings.append(f"{path.relative_to(ROOT)}:{line}: {label}") + assert findings == [], "\n".join(findings) + + +def test_byok_encryption_is_randomized_authenticated_and_tamper_evident(monkeypatch): + monkeypatch.setenv("SCM_CLOUD_SECRET_KEY", "qualification-secret-key-with-at-least-32-chars") + plaintext = "provider-token-that-must-remain-secret" + first = accounts._encrypt(plaintext) + second = accounts._encrypt(plaintext) + + assert first.startswith("v1:") + assert first != second + assert accounts._decrypt(first) == plaintext + + replacement = "A" if first[-1] != "A" else "B" + tampered = first[:-1] + replacement + with pytest.raises((InvalidTag, ValueError)): + accounts._decrypt(tampered) + + +def test_wildcard_cors_never_allows_browser_credentials(): + from src.api.main import app + + cors = next( + middleware + for middleware in app.user_middleware + if middleware.cls.__name__ == "CORSMiddleware" + ) + assert cors.kwargs["allow_origins"] == ["*"] + assert cors.kwargs["allow_credentials"] is False + + +def test_invalid_bearer_is_not_echoed_in_error(product_client, monkeypatch): + token = "scm_live_attacker_supplied_secret_material" + monkeypatch.setenv("SCM_CLOUD_AUTH", "1") + response = product_client.post( + "/v1/memories", + headers={"Authorization": f"Bearer {token}"}, + json={"text": "hello"}, + ) + assert response.status_code == 401 + assert token not in response.text + + +def test_cli_health_output_reports_presence_without_printing_credentials(tmp_path): + secret = "qualification-secret-never-print-this" + env = os.environ.copy() + env.update( + { + "OPENAI_API_KEY": secret, + "SCM_DATA_DIR": str(tmp_path / "doctor"), + "LLM_PROVIDER": "", + "SCM_EMBEDDING_BACKEND": "hash", + } + ) + result = subprocess.run( + [sys.executable, "-m", "src.cli.main", "doctor", "--json"], + cwd=ROOT, + env=env, + text=True, + capture_output=True, + timeout=60, + check=False, + ) + assert result.returncode == 0, result.stderr + assert secret not in result.stdout + assert '"openai": true' in result.stdout.lower() diff --git a/tests/test_ab_hierarchical.py b/tests/test_ab_hierarchical.py index 3fa8a17..f4368bd 100644 --- a/tests/test_ab_hierarchical.py +++ b/tests/test_ab_hierarchical.py @@ -32,7 +32,7 @@ memory system for AI agents inspired by how the brain consolidates memories during sleep.""" -def test_mode(name, text, hierarchical): +def run_mode(name, text, hierarchical): os.environ["HIERARCHICAL_EXTRACTION"] = "true" if hierarchical else "false" # Reload config import importlib @@ -63,8 +63,8 @@ def main(): print("=" * 60) # Multi-topic - flat_c, flat_t = test_mode("MULTI-TOPIC INPUT", MULTI_TOPIC, hierarchical=False) - hier_c, hier_t = test_mode("MULTI-TOPIC INPUT", MULTI_TOPIC, hierarchical=True) + flat_c, flat_t = run_mode("MULTI-TOPIC INPUT", MULTI_TOPIC, hierarchical=False) + hier_c, hier_t = run_mode("MULTI-TOPIC INPUT", MULTI_TOPIC, hierarchical=True) print(f"\n{'='*60}") print(f"MULTI-TOPIC COMPARISON") @@ -75,8 +75,8 @@ def main(): print(f"Time cost: +{hier_t - flat_t:.2f}s ({(hier_t/flat_t - 1)*100:+.0f}%)") # Long narrative - flat_c2, flat_t2 = test_mode("LONG NARRATIVE", LONG_NARRATIVE, hierarchical=False) - hier_c2, hier_t2 = test_mode("LONG NARRATIVE", LONG_NARRATIVE, hierarchical=True) + flat_c2, flat_t2 = run_mode("LONG NARRATIVE", LONG_NARRATIVE, hierarchical=False) + hier_c2, hier_t2 = run_mode("LONG NARRATIVE", LONG_NARRATIVE, hierarchical=True) print(f"\n{'='*60}") print(f"LONG NARRATIVE COMPARISON") diff --git a/tests/test_crazy_brutal.py b/tests/test_crazy_brutal.py index 479eb20..9bd7213 100644 --- a/tests/test_crazy_brutal.py +++ b/tests/test_crazy_brutal.py @@ -889,11 +889,12 @@ def test_very_long_descriptions(self): def test_special_characters_in_content(self): """Special characters and unicode""" orchestrator = SleepCycleOrchestrator() + special_quote = '"' concepts = [Concept( id=f"special_c{i}", type=ConceptType.FACT, - description=f"Special chars: !@#$%^&*() 你好 😂 {'\"'} \\n\\t {i}", + description=f"Special chars: !@#$%^&*() 你好 😂 {special_quote} \\n\\t {i}", importance=ImportanceVector(), strength=1.0 ) for i in range(30)] @@ -978,4 +979,4 @@ def run_crazy_tests(): if __name__ == '__main__': success = run_crazy_tests() - sys.exit(0 if success else 1) \ No newline at end of file + sys.exit(0 if success else 1) diff --git a/tests/test_wake_summary_e2e.py b/tests/test_wake_summary_e2e.py index 3458fef..7ca025f 100644 --- a/tests/test_wake_summary_e2e.py +++ b/tests/test_wake_summary_e2e.py @@ -10,6 +10,15 @@ import tempfile from pathlib import Path +import pytest + + +if os.environ.get("SCM_RUN_PAID_E2E") != "1": + pytest.skip( + "paid DeepSeek/Ollama E2E is opt-in; set SCM_RUN_PAID_E2E=1 to run it", + allow_module_level=True, + ) + # Force a clean data dir so we test cold-start behavior _DATA_DIR = tempfile.mkdtemp(prefix="scm_e2e_") os.environ["SCM_DATA_DIR"] = _DATA_DIR @@ -17,11 +26,8 @@ os.environ["SCM_EMBEDDING_MODEL"] = "nomic-embed-text" os.environ["LLM_PROVIDER"] = "deepseek" -# Make sure DeepSeek key is loaded -from dotenv import load_dotenv -load_dotenv() - -assert os.environ.get("DEEPSEEK_API_KEY"), "DEEPSEEK_API_KEY missing from .env" +if not os.environ.get("DEEPSEEK_API_KEY"): + pytest.skip("SCM_RUN_PAID_E2E=1 requires DEEPSEEK_API_KEY", allow_module_level=True) print(f"=== Cold-start E2E test ===") print(f"Data dir: {_DATA_DIR}")