docs: add memory activation gate checklist

docs/security/user-data-isolation-activation-gate.md

@@ -0,0 +1,77 @@
+# User Data Isolation Activation Gate
+
+## Purpose
+
+Define the final conditions required before enabling memory search or cross-agent memory recall.
+
+## Current State
+
+```text
+memory search: disabled
+isolation gate: enforced
+```
+
+## Activation Checklist
+
+Memory search may be enabled only when all checks pass.
+
+### Storage isolation
+
+- [ ] Records are stored under scoped namespaces.
+- [ ] No global shared memory path is used for production search.
+- [ ] Every record has required scope metadata.
+
+### Read isolation
+
+- [ ] Scope envelope is attached to every search request.
+- [ ] Provider filters before ranking.
+- [ ] Out-of-scope records never reach summarization.
+- [ ] Provenance is returned for every accepted record.
+
+### Audit
+
+- [ ] Isolation audit reports `healthy`.
+- [ ] Denied records count is zero.
+- [ ] Invalid records count is zero.
+- [ ] Safe-mode path is tested.
+
+### Provider
+
+Required contract:
+
+```json
+{
+  "scope_filter_before_ranking": true,
+  "provenance_required": true,
+  "allow_global_index": false
+}
+```
+
+### Owner Gate
+
+Activation requires explicit approval because memory/search changes the data exposure boundary.
+
+## Failure Behavior
+
+If any check fails:
+
+```text
+memory remains disabled
+        +
+audit event generated
+        +
+safe-mode available for autonomous operations
+```
+
+## Non-Goals
+
+This gate does not:
+
+- grant RBAC permissions;
+- replace application authorization;
+- allow global recall for convenience;
+- bypass tenant/user/project boundaries.
+
+## Release Rule
+
+User-data isolation is a release blocker, not an optimization item.

docs/security/user-data-isolation-deployment-runbook.md

@@ -0,0 +1,112 @@
+# User Data Isolation Deployment Runbook
+
+## Goal
+
+Deploy user-data isolation controls before enabling memory search or cross-agent recall.
+
+## Current Safety State
+
+Default:
+
+```text
+AIOS_MEMORY_ENABLED=false
+```
+
+Memory search remains disabled until isolation acceptance passes.
+
+## Deployment Components
+
+1. Memory scope guard
+
+```text
+scripts/security/memory_scope_guard.py
+```
+
+2. Provider gate
+
+```text
+scripts/security/memory_provider_gate.py
+```
+
+3. Isolation audit service
+
+```text
+user-data-isolation-audit.service
+user-data-isolation-audit.timer
+```
+
+4. Activation gate
+
+```text
+scripts/security/memory_search_activation_check.py
+```
+
+## Deployment Order
+
+```text
+Install audit
+        ↓
+Run isolation scan
+        ↓
+Fix unscoped records
+        ↓
+Enable provider scope contract
+        ↓
+Run acceptance suite
+        ↓
+Evaluate activation gate
+        ↓
+Only then enable memory search
+```
+
+## Acceptance Requirements
+
+Before enabling memory search:
+
+### Isolation
+
+- no untagged records;
+- no cross-user records;
+- no cross-project records;
+- no cross-tenant records;
+- no global shared memory paths;
+- all records have provenance.
+
+### Provider
+
+Required:
+
+```json
+{
+  "scope_filter_before_ranking": true,
+  "provenance_required": true,
+  "allow_global_index": false
+}
+```
+
+### Activation Gate
+
+The activation check must return:
+
+```json
+{
+  "status": "allowed"
+}
+```
+
+Anything else keeps memory disabled.
+
+## Incident Response
+
+If contamination is detected:
+
+1. Disable memory/search provider.
+2. Enter autonomous safe mode if operational state may be affected.
+3. Preserve audit evidence.
+4. Identify scope violation source.
+5. Migrate records into scoped namespaces.
+6. Rerun acceptance before reactivation.
+
+## Redline
+
+Convenience, recall quality, and search coverage never override user-data isolation.

ops/ci/run-user-data-isolation-acceptance.sh

@@ -0,0 +1,58 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+cd "$ROOT"
+
+python3 -m pytest \
+  tests/security/test_memory_scope_guard.py \
+  tests/security/test_memory_provider_gate.py
+
+TMP_DIR="$(mktemp -d)"
+trap 'rm -rf "$TMP_DIR"' EXIT
+
+cat > "$TMP_DIR/report.json" <<'JSON'
+{
+  "status": "healthy",
+  "counts": {
+    "denied": 0,
+    "invalid": 0
+  }
+}
+JSON
+
+cat > "$TMP_DIR/provider.json" <<'JSON'
+{
+  "name": "ci-provider",
+  "scope_filter_before_ranking": true,
+  "provenance_required": true,
+  "fallback_mode": "text-only",
+  "allow_global_index": false
+}
+JSON
+
+python3 scripts/security/memory_search_activation_check.py \
+  --isolation-report "$TMP_DIR/report.json" \
+  --provider-config "$TMP_DIR/provider.json" \
+  --request-enable
+
+# The following must stay blocked: unsafe provider configuration.
+cat > "$TMP_DIR/provider-unsafe.json" <<'JSON'
+{
+  "name": "unsafe-provider",
+  "scope_filter_before_ranking": false,
+  "provenance_required": false,
+  "fallback_mode": "global-search",
+  "allow_global_index": true
+}
+JSON
+
+if python3 scripts/security/memory_search_activation_check.py \
+  --isolation-report "$TMP_DIR/report.json" \
+  --provider-config "$TMP_DIR/provider-unsafe.json" \
+  --request-enable; then
+  echo "ERROR: unsafe memory provider activation was allowed" >&2
+  exit 1
+fi
+
+echo "user-data-isolation acceptance: PASS"

ops/ci/user-data-isolation-gate.yaml

@@ -0,0 +1,26 @@
+name: user-data-isolation-gate
+
+purpose: >-
+  Block memory/search activation unless user-data isolation controls are healthy.
+
+checks:
+  - name: memory-scope-tests
+    command: bash ops/ci/run-user-data-isolation-acceptance.sh
+
+  - name: activation-policy
+    requirements:
+      isolation_report: healthy
+      denied_records: 0
+      invalid_records: 0
+      scope_filter_before_ranking: true
+      provenance_required: true
+      allow_global_index: false
+
+failure_policy:
+  action: block
+  reason: user-data-isolation-redline
+
+notes:
+  - memory search must remain disabled when this gate fails.
+  - no provider may bypass scope validation.
+  - passing this gate does not grant access; RBAC and owner policy still apply.

ops/env/memory-provider-scope-contract.json.example

@@ -0,0 +1,21 @@
+{
+  "name": "memory-provider-default",
+  "scope_filter_before_ranking": true,
+  "provenance_required": true,
+  "fallback_mode": "text-only",
+  "allow_global_index": false,
+  "allowed_record_fields": [
+    "record_id",
+    "user_id",
+    "tenant_id",
+    "group_id",
+    "project_id",
+    "workspace_id",
+    "session_id",
+    "branch",
+    "baseline",
+    "agent_id",
+    "source_tool"
+  ],
+  "notes": "Memory search cannot be enabled unless this contract and isolation audit are healthy."
+}

ops/env/memory-search-activation-check.env.example

@@ -0,0 +1,11 @@
+# Memory search activation gate configuration
+
+AIOS_MEMORY_ACTIVATION_CHECK_SCRIPT=/data/aios-gitops/workspace/scripts/security/memory_search_activation_check.py
+AIOS_MEMORY_SCOPE_REPORT=/data/aios-gitops/runtime/state/memory-scope-guard-report.json
+AIOS_MEMORY_PROVIDER_CONFIG=/etc/aios/memory-provider-scope-contract.json
+
+# Keep empty until explicit owner-approved activation.
+AIOS_MEMORY_ENABLE_REQUEST_FLAG=
+
+# Safe default:
+AIOS_MEMORY_ENABLED=false

ops/env/user-data-isolation-audit.env.example

@@ -0,0 +1,10 @@
+# User data isolation audit configuration
+
+AIOS_MEMORY_GUARD_SCRIPT=/data/aios-gitops/workspace/scripts/security/memory_scope_guard.py
+AIOS_MEMORY_ROOT=/data/aios-gitops/runtime/memory
+AIOS_MEMORY_SCOPE_REPORT=/data/aios-gitops/runtime/state/memory-scope-guard-report.json
+AUTONOMOUS_SAFE_MODE_FILE=/data/aios-gitops/runtime/state/autonomous-safe-mode.json
+
+# Memory search remains disabled until isolation acceptance passes.
+AIOS_MEMORY_ENABLED=false
+AIOS_MEMORY_REQUIRE_SCOPE=true

ops/systemd/memory-search-activation-check.service

@@ -0,0 +1,21 @@
+[Unit]
+Description=Memory search activation safety gate
+Documentation=file:docs/security/user-data-isolation-redline.md
+Documentation=file:docs/security/user-data-isolation-deployment-runbook.md
+After=user-data-isolation-audit.service
+
+[Service]
+Type=oneshot
+EnvironmentFile=-/etc/aios/memory-search-activation-check.env
+ExecStart=/bin/sh -lc 'exec /usr/bin/python3 ${AIOS_MEMORY_ACTIVATION_CHECK_SCRIPT} --isolation-report ${AIOS_MEMORY_SCOPE_REPORT} --provider-config ${AIOS_MEMORY_PROVIDER_CONFIG} ${AIOS_MEMORY_ENABLE_REQUEST_FLAG:-}'
+TimeoutStartSec=60
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectHome=true
+ProtectSystem=full
+ReadOnlyPaths=/data/aios-gitops/runtime/state
+ReadOnlyPaths=/etc/aios
+
+# Non-zero exit means activation is blocked. This is expected when memory search
+# has not been explicitly requested or isolation is not fully healthy.
+SuccessExitStatus=78

ops/systemd/user-data-isolation-audit.service

@@ -0,0 +1,17 @@
+[Unit]
+Description=User data isolation audit guard
+Documentation=file:docs/security/user-data-isolation-redline.md
+Wants=network-online.target
+After=network-online.target
+
+[Service]
+Type=oneshot
+EnvironmentFile=-/etc/aios/user-data-isolation-audit.env
+ExecStart=/bin/sh -lc 'exec /usr/bin/python3 ${AIOS_MEMORY_GUARD_SCRIPT} audit --emit-safe-mode --json'
+TimeoutStartSec=120
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectHome=true
+ProtectSystem=full
+ReadWritePaths=/data/aios-gitops/runtime/state
+ReadOnlyPaths=/data/aios-gitops/runtime/memory

ops/systemd/user-data-isolation-audit.timer

@@ -0,0 +1,11 @@
+[Unit]
+Description=Periodic user data isolation audit
+
+[Timer]
+OnBootSec=5min
+OnUnitActiveSec=5min
+RandomizedDelaySec=30s
+Persistent=true
+
+[Install]
+WantedBy=timers.target

scripts/security/memory_provider_gate.py

@@ -0,0 +1,74 @@
+#!/usr/bin/env python3
+"""Provider boundary for memory/search results.
+
+The provider gate runs before ranking or summarization. It accepts candidate
+records only after scope validation. It never returns unscoped content.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+from pathlib import Path
+import sys
+
+from scripts.security.memory_scope_guard import record_matches_scope
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--scope-file", required=True, type=Path)
+    parser.add_argument("--candidates-file", required=True, type=Path)
+    parser.add_argument("--output-file", required=True, type=Path)
+    args = parser.parse_args()
+
+    scope = json.loads(args.scope_file.read_text(encoding="utf-8"))
+    candidates = json.loads(args.candidates_file.read_text(encoding="utf-8"))
+
+    if not isinstance(candidates, list):
+        print("candidates must be a JSON list", file=sys.stderr)
+        return 3
+
+    allowed = []
+    denied = []
+
+    for candidate in candidates:
+        if not isinstance(candidate, dict):
+            denied.append({"reason": "invalid_candidate_type"})
+            continue
+
+        metadata = candidate.get("metadata", {})
+        ok, reasons = record_matches_scope(metadata, scope)
+
+        # Never forward the body/content of denied records.
+        if ok:
+            allowed.append({
+                "record_id": metadata.get("record_id"),
+                "metadata": metadata,
+                "provenance": candidate.get("provenance", {}),
+            })
+        else:
+            denied.append({
+                "record_id": metadata.get("record_id"),
+                "reasons": reasons,
+            })
+
+    result = {
+        "status": "healthy" if not denied else "filtered",
+        "allowed_count": len(allowed),
+        "denied_count": len(denied),
+        "allowed": allowed,
+        "denied": denied,
+    }
+
+    args.output_file.parent.mkdir(parents=True, exist_ok=True)
+    args.output_file.write_text(
+        json.dumps(result, ensure_ascii=False, indent=2) + "\n",
+        encoding="utf-8",
+    )
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/security/memory_scope_guard.py

@@ -0,0 +1,477 @@
+#!/usr/bin/env python3
+"""User data isolation guard for memory/search providers.
+
+This guard enforces scope-first access for memory/state records. It is designed
+for two production-safe uses:
+
+1. Audit memory directories for unscoped or cross-scope records.
+2. Filter provider candidates before ranking/summarization receives them.
+
+It never prints record content. Reports contain only provenance, metadata, and
+violation reasons.
+"""
+
+from __future__ import annotations
+
+import argparse
+import datetime as _dt
+import json
+import os
+from pathlib import Path
+import re
+import sys
+import tempfile
+from typing import Any
+
+REQUIRED_RECORD_FIELDS = (
+    "record_id",
+    "user_id",
+    "project_id",
+    "workspace_id",
+    "session_id",
+    "branch",
+    "baseline",
+    "agent_id",
+    "source_tool",
+    "created_at",
+    "schema_version",
+    "sensitivity",
+    "retention_class",
+)
+
+# tenant_id and group_id are deployment-dependent. At least one must exist.
+TENANT_FIELDS = ("tenant_id", "group_id")
+
+REQUIRED_SCOPE_FIELDS = (
+    "request_user_id",
+    "request_project_id",
+    "request_workspace_id",
+    "request_session_id",
+    "current_branch",
+    "current_baseline",
+    "purpose",
+)
+
+TENANT_SCOPE_FIELDS = ("request_tenant_id", "request_group_id")
+
+TEXT_EXTENSIONS = {".json", ".jsonl", ".md", ".txt"}
+DENY_EXIT = 2
+CONFIG_EXIT = 3
+
+
+def utc_now() -> str:
+    return _dt.datetime.now(_dt.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def atomic_write_json(path: Path, payload: dict[str, Any], mode: int = 0o644) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=str(path.parent))
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=False, indent=2, sort_keys=True)
+            handle.write("\n")
+        os.chmod(tmp_name, mode)
+        os.replace(tmp_name, path)
+    finally:
+        try:
+            os.unlink(tmp_name)
+        except FileNotFoundError:
+            pass
+
+
+def read_json(path: Path) -> Any:
+    with path.open("r", encoding="utf-8") as handle:
+        return json.load(handle)
+
+
+def load_scope(path: Path | None) -> dict[str, Any]:
+    if path is not None:
+        scope = read_json(path)
+    else:
+        raw = os.environ.get("AIOS_SCOPE_ENVELOPE_JSON", "").strip()
+        if not raw:
+            raise ValueError("scope envelope is required via --scope-file or AIOS_SCOPE_ENVELOPE_JSON")
+        scope = json.loads(raw)
+
+    if not isinstance(scope, dict):
+        raise ValueError("scope envelope must be a JSON object")
+    return scope
+
+
+def parse_simple_frontmatter(text: str) -> dict[str, Any]:
+    """Parse a conservative YAML-like frontmatter block.
+
+    Supported format:
+    ---
+    key: value
+    nested.key: value
+    tags: [a, b]
+    ---
+
+    This avoids a YAML dependency and intentionally ignores complex YAML.
+    """
+
+    if not text.startswith("---\n"):
+        return {}
+
+    end = text.find("\n---", 4)
+    if end == -1:
+        return {}
+
+    block = text[4:end]
+    result: dict[str, Any] = {}
+    for line in block.splitlines():
+        line = line.strip()
+        if not line or line.startswith("#") or ":" not in line:
+            continue
+        key, value = line.split(":", 1)
+        key = key.strip()
+        value = value.strip().strip('"').strip("'")
+        if value.startswith("[") and value.endswith("]"):
+            items = [item.strip().strip('"').strip("'") for item in value[1:-1].split(",") if item.strip()]
+            set_nested(result, key, items)
+        else:
+            set_nested(result, key, value)
+    return result
+
+
+def set_nested(target: dict[str, Any], dotted_key: str, value: Any) -> None:
+    parts = [part for part in dotted_key.split(".") if part]
+    if not parts:
+        return
+    current = target
+    for part in parts[:-1]:
+        child = current.get(part)
+        if not isinstance(child, dict):
+            child = {}
+            current[part] = child
+        current = child
+    current[parts[-1]] = value
+
+
+def metadata_from_payload(payload: Any) -> dict[str, Any]:
+    if not isinstance(payload, dict):
+        return {}
+
+    for key in ("metadata", "scope", "record_scope"):
+        value = payload.get(key)
+        if isinstance(value, dict):
+            return value
+
+    # Some legacy records store scope fields at top level. Accept only metadata,
+    # not record body content, and never use body content as proof of scope.
+    candidate = {key: payload[key] for key in REQUIRED_RECORD_FIELDS if key in payload}
+    for key in TENANT_FIELDS:
+        if key in payload:
+            candidate[key] = payload[key]
+    return candidate
+
+
+def load_record_metadata(path: Path) -> tuple[dict[str, Any], dict[str, Any]]:
+    provenance = {
+        "path": str(path),
+        "suffix": path.suffix.lower(),
+        "mtime": _dt.datetime.fromtimestamp(path.stat().st_mtime, _dt.timezone.utc).isoformat().replace("+00:00", "Z"),
+    }
+
+    suffix = path.suffix.lower()
+    if suffix == ".json":
+        payload = read_json(path)
+        return metadata_from_payload(payload), provenance
+
+    if suffix == ".jsonl":
+        metadata: dict[str, Any] = {}
+        with path.open("r", encoding="utf-8") as handle:
+            for line in handle:
+                line = line.strip()
+                if not line:
+                    continue
+                payload = json.loads(line)
+                metadata = metadata_from_payload(payload)
+                break
+        return metadata, provenance
+
+    if suffix in {".md", ".txt"}:
+        text = path.read_text(encoding="utf-8", errors="replace")
+        frontmatter = parse_simple_frontmatter(text)
+        return metadata_from_payload(frontmatter) or frontmatter, provenance
+
+    return {}, provenance
+
+
+def list_candidate_files(root: Path) -> list[Path]:
+    if root.is_file():
+        return [root]
+    if not root.exists():
+        return []
+    files: list[Path] = []
+    for path in root.rglob("*"):
+        if path.is_file() and path.suffix.lower() in TEXT_EXTENSIONS:
+            files.append(path)
+    return sorted(files)
+
+
+def missing_fields(metadata: dict[str, Any]) -> list[str]:
+    missing = [field for field in REQUIRED_RECORD_FIELDS if metadata.get(field) in (None, "")]
+    if not any(metadata.get(field) not in (None, "") for field in TENANT_FIELDS):
+        missing.append("tenant_id_or_group_id")
+    return missing
+
+
+def validate_scope_envelope(scope: dict[str, Any]) -> list[str]:
+    missing = [field for field in REQUIRED_SCOPE_FIELDS if scope.get(field) in (None, "")]
+    if not any(scope.get(field) not in (None, "") for field in TENANT_SCOPE_FIELDS):
+        missing.append("request_tenant_id_or_request_group_id")
+    return missing
+
+
+def as_set(value: Any) -> set[str]:
+    if value is None or value == "":
+        return set()
+    if isinstance(value, list):
+        return {str(item) for item in value if item not in (None, "")}
+    if isinstance(value, tuple) or isinstance(value, set):
+        return {str(item) for item in value if item not in (None, "")}
+    return {str(value)}
+
+
+def scope_value(scope: dict[str, Any], primary: str, allowed: str | None = None) -> set[str]:
+    values = as_set(scope.get(primary))
+    if allowed:
+        values |= as_set(scope.get(allowed))
+    return values
+
+
+def record_matches_scope(metadata: dict[str, Any], scope: dict[str, Any]) -> tuple[bool, list[str]]:
+    reasons: list[str] = []
+
+    missing = missing_fields(metadata)
+    if missing:
+        return False, [f"missing_metadata:{','.join(missing)}"]
+
+    scope_missing = validate_scope_envelope(scope)
+    if scope_missing:
+        return False, [f"invalid_scope:{','.join(scope_missing)}"]
+
+    def deny_if_mismatch(record_key: str, scope_key: str, reason: str, allowed_key: str | None = None) -> None:
+        allowed = scope_value(scope, scope_key, allowed_key)
+        if str(metadata.get(record_key)) not in allowed:
+            reasons.append(reason)
+
+    # Tenant/group policy: if both sides have tenant_id, compare tenant_id. If not,
+    # use group_id. Cross-tenant/group is always denied by default.
+    if metadata.get("tenant_id") and scope.get("request_tenant_id"):
+        deny_if_mismatch("tenant_id", "request_tenant_id", "tenant_mismatch")
+    elif metadata.get("group_id") and scope.get("request_group_id"):
+        deny_if_mismatch("group_id", "request_group_id", "group_mismatch")
+    else:
+        reasons.append("tenant_or_group_scope_unprovable")
+
+    deny_if_mismatch("user_id", "request_user_id", "user_mismatch", "allowed_user_ids")
+    deny_if_mismatch("project_id", "request_project_id", "project_mismatch", "allowed_project_ids")
+    deny_if_mismatch("workspace_id", "request_workspace_id", "workspace_mismatch", "allowed_workspace_ids")
+
+    session_policy = str(scope.get("session_policy", "same_session")).lower()
+    if session_policy == "same_session":
+        deny_if_mismatch("session_id", "request_session_id", "session_mismatch")
+    elif session_policy == "allow_project_sessions":
+        # Same project/workspace checks above still apply. Do not add session reason.
+        pass
+    else:
+        reasons.append(f"unsupported_session_policy:{session_policy}")
+
+    current_branch = scope_value(scope, "current_branch", "allowed_branches")
+    if str(metadata.get("branch")) not in current_branch:
+        reasons.append("branch_mismatch")
+
+    current_baseline = scope_value(scope, "current_baseline", "allowed_baselines")
+    if str(metadata.get("baseline")) not in current_baseline:
+        reasons.append("baseline_mismatch")
+
+    allowed_agents = scope_value(scope, "request_agent_id", "allowed_agent_ids")
+    if allowed_agents and str(metadata.get("agent_id")) not in allowed_agents:
+        reasons.append("agent_mismatch")
+
+    return not reasons, reasons
+
+
+def path_namespace_violations(path: Path, metadata: dict[str, Any], root: Path) -> list[str]:
+    """Detect obvious global/shared path misuse.
+
+    This is a supplemental safety check. It does not grant access; it can only
+    add violations.
+    """
+
+    try:
+        relative = path.relative_to(root)
+    except ValueError:
+        return []
+
+    parts = relative.parts
+    if not parts:
+        return []
+
+    lower_parts = [part.lower() for part in parts]
+    violations: list[str] = []
+
+    # Disallow records directly under a global memory root.
+    if len(parts) <= 2:
+        violations.append("global_memory_path")
+
+    # If path includes explicit namespace words, validate their following value.
+    checks = {
+        "tenant": metadata.get("tenant_id"),
+        "tenant_id": metadata.get("tenant_id"),
+        "group": metadata.get("group_id"),
+        "group_id": metadata.get("group_id"),
+        "user": metadata.get("user_id"),
+        "user_id": metadata.get("user_id"),
+        "project": metadata.get("project_id"),
+        "project_id": metadata.get("project_id"),
+        "session": metadata.get("session_id"),
+        "session_id": metadata.get("session_id"),
+    }
+    for key, expected in checks.items():
+        if key in lower_parts:
+            index = lower_parts.index(key)
+            if index + 1 < len(parts) and expected and str(expected) != parts[index + 1]:
+                violations.append(f"path_metadata_mismatch:{key}")
+
+    return violations
+
+
+def scan(root: Path, scope: dict[str, Any] | None, require_scope: bool) -> dict[str, Any]:
+    files = list_candidate_files(root)
+    allowed: list[dict[str, Any]] = []
+    denied: list[dict[str, Any]] = []
+    invalid: list[dict[str, Any]] = []
+
+    for path in files:
+        try:
+            metadata, provenance = load_record_metadata(path)
+        except Exception as exc:  # noqa: BLE001 - report diagnostics without content
+            invalid.append({"path": str(path), "reason": f"parse_error:{type(exc).__name__}", "message": str(exc)[:240]})
+            continue
+
+        reasons: list[str] = []
+        missing = missing_fields(metadata)
+        if missing:
+            reasons.append(f"missing_metadata:{','.join(missing)}")
+
+        reasons.extend(path_namespace_violations(path, metadata, root))
+
+        if require_scope:
+            if scope is None:
+                reasons.append("scope_required")
+            else:
+                matched, match_reasons = record_matches_scope(metadata, scope)
+                if not matched:
+                    reasons.extend(match_reasons)
+
+        record = {
+            "record_id": metadata.get("record_id"),
+            "metadata": {key: metadata.get(key) for key in sorted(set(REQUIRED_RECORD_FIELDS + TENANT_FIELDS)) if key in metadata},
+            "provenance": provenance,
+        }
+
+        if reasons:
+            record["reasons"] = sorted(set(reasons))
+            denied.append(record)
+        else:
+            allowed.append(record)
+
+    status = "healthy" if not denied and not invalid else "unhealthy"
+    return {
+        "service": "memory-scope-guard",
+        "status": status,
+        "generated_at": utc_now(),
+        "root": str(root),
+        "counts": {
+            "files": len(files),
+            "allowed": len(allowed),
+            "denied": len(denied),
+            "invalid": len(invalid),
+        },
+        "allowed": allowed,
+        "denied": denied,
+        "invalid": invalid,
+    }
+
+
+def write_safe_mode(report: dict[str, Any], path: Path) -> None:
+    marker = {
+        "service": "autonomous-safe-mode",
+        "status": "active",
+        "entered_at": utc_now(),
+        "reason": "user_data_isolation_violation",
+        "source": "memory-scope-guard",
+        "report_file": report.get("report_file"),
+        "counts": report.get("counts", {}),
+        "operator_action": (
+            "Keep memory/search disabled, inspect denied/invalid records, migrate to scoped namespaces, "
+            "then rerun isolation acceptance before enabling memory search."
+        ),
+    }
+    atomic_write_json(path, marker)
+
+
+def parse_args(argv: list[str]) -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Audit or filter memory records with user-data isolation enforcement.")
+    parser.add_argument("--root", default=os.environ.get("AIOS_MEMORY_ROOT", "/data/aios-gitops/runtime/memory"))
+    parser.add_argument("--scope-file", type=Path, default=None)
+    parser.add_argument("--report-file", type=Path, default=Path(os.environ.get("AIOS_MEMORY_SCOPE_REPORT", "/data/aios-gitops/runtime/state/memory-scope-guard-report.json")))
+    parser.add_argument("--safe-mode-file", type=Path, default=Path(os.environ.get("AUTONOMOUS_SAFE_MODE_FILE", "/data/aios-gitops/runtime/state/autonomous-safe-mode.json")))
+    parser.add_argument("--require-scope", action="store_true", default=os.environ.get("AIOS_MEMORY_REQUIRE_SCOPE", "true").lower() not in {"0", "false", "no"})
+    parser.add_argument("--emit-safe-mode", action="store_true")
+    parser.add_argument("--json", action="store_true")
+    parser.add_argument("command", choices=["audit", "filter"], nargs="?", default="audit")
+    return parser.parse_args(argv)
+
+
+def main(argv: list[str]) -> int:
+    args = parse_args(argv)
+    scope = None
+    if args.require_scope:
+        try:
+            scope = load_scope(args.scope_file)
+        except Exception as exc:  # noqa: BLE001
+            report = {
+                "service": "memory-scope-guard",
+                "status": "unhealthy",
+                "generated_at": utc_now(),
+                "error": f"scope_error:{type(exc).__name__}",
+                "message": str(exc),
+                "counts": {"files": 0, "allowed": 0, "denied": 0, "invalid": 1},
+            }
+            report["report_file"] = str(args.report_file)
+            atomic_write_json(args.report_file, report)
+            if args.emit_safe_mode:
+                write_safe_mode(report, args.safe_mode_file)
+            json.dump(report, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
+            sys.stdout.write("\n")
+            return CONFIG_EXIT
+
+    report = scan(Path(args.root), scope, args.require_scope)
+    report["report_file"] = str(args.report_file)
+    atomic_write_json(args.report_file, report)
+
+    if report["status"] != "healthy" and args.emit_safe_mode:
+        write_safe_mode(report, args.safe_mode_file)
+
+    if args.json or report["status"] != "healthy":
+        json.dump(report, sys.stdout, ensure_ascii=False, indent=2, sort_keys=True)
+        sys.stdout.write("\n")
+    else:
+        print(
+            "healthy: "
+            f"files={report['counts']['files']} "
+            f"allowed={report['counts']['allowed']} "
+            f"denied={report['counts']['denied']} "
+            f"invalid={report['counts']['invalid']}"
+        )
+
+    return 0 if report["status"] == "healthy" else DENY_EXIT
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv[1:]))

scripts/security/memory_search_activation_check.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""Activation gate for memory/search providers.
+
+Memory search may be enabled only after isolation audit is healthy and the
+activation config explicitly requests scoped mode. This script is intentionally
+conservative: missing reports, denied records, invalid records, or unsupported
+provider modes block activation.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+import sys
+from typing import Any
+
+BLOCKED_EXIT = 78
+CONFIG_EXIT = 3
+
+
+def read_json(path: Path) -> dict[str, Any]:
+    with path.open("r", encoding="utf-8") as handle:
+        value = json.load(handle)
+    if not isinstance(value, dict):
+        raise ValueError(f"expected JSON object: {path}")
+    return value
+
+
+def env_bool(name: str, default: bool = False) -> bool:
+    raw = os.environ.get(name)
+    if raw is None or raw == "":
+        return default
+    return raw.strip().lower() in {"1", "true", "yes", "y", "on"}
+
+
+def evaluate(report: dict[str, Any], provider: dict[str, Any], requested_enable: bool) -> dict[str, Any]:
+    reasons: list[str] = []
+
+    if not requested_enable:
+        reasons.append("memory_search_not_requested")
+
+    if report.get("status") != "healthy":
+        reasons.append("isolation_report_not_healthy")
+
+    counts = report.get("counts", {})
+    if counts.get("denied", 0):
+        reasons.append("denied_records_present")
+    if counts.get("invalid", 0):
+        reasons.append("invalid_records_present")
+
+    if not provider.get("scope_filter_before_ranking", False):
+        reasons.append("provider_does_not_filter_before_ranking")
+    if not provider.get("provenance_required", False):
+        reasons.append("provider_does_not_return_provenance")
+    if provider.get("fallback_mode") not in {"text-only", "no-memory", "scoped-only"}:
+        reasons.append("provider_fallback_mode_unsafe")
+    if provider.get("allow_global_index", False):
+        reasons.append("global_index_not_allowed")
+
+    status = "allowed" if requested_enable and not reasons else "blocked"
+    return {
+        "service": "memory-search-activation-check",
+        "status": status,
+        "reasons": reasons,
+        "report_counts": counts,
+        "provider": {
+            "name": provider.get("name", "unknown"),
+            "scope_filter_before_ranking": provider.get("scope_filter_before_ranking", False),
+            "provenance_required": provider.get("provenance_required", False),
+            "fallback_mode": provider.get("fallback_mode"),
+            "allow_global_index": provider.get("allow_global_index", False),
+        },
+    }
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Check whether memory search may be enabled safely.")
+    parser.add_argument(
+        "--isolation-report",
+        type=Path,
+        default=Path(os.environ.get("AIOS_MEMORY_SCOPE_REPORT", "/data/aios-gitops/runtime/state/memory-scope-guard-report.json")),
+    )
+    parser.add_argument(
+        "--provider-config",
+        type=Path,
+        default=Path(os.environ.get("AIOS_MEMORY_PROVIDER_CONFIG", "/etc/aios/memory-provider-scope-contract.json")),
+    )
+    parser.add_argument("--request-enable", action="store_true", default=env_bool("AIOS_MEMORY_ENABLE_REQUESTED", False))
+    parser.add_argument("--output-file", type=Path, default=None)
+    args = parser.parse_args()
+
+    try:
+        report = read_json(args.isolation_report)
+        provider = read_json(args.provider_config)
+    except Exception as exc:  # noqa: BLE001
+        result = {
+            "service": "memory-search-activation-check",
+            "status": "blocked",
+            "reasons": [f"configuration_error:{type(exc).__name__}"],
+            "message": str(exc),
+        }
+        print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
+        return CONFIG_EXIT
+
+    result = evaluate(report, provider, args.request_enable)
+    text = json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True) + "\n"
+    if args.output_file:
+        args.output_file.parent.mkdir(parents=True, exist_ok=True)
+        args.output_file.write_text(text, encoding="utf-8")
+    print(text, end="")
+    return 0 if result["status"] == "allowed" else BLOCKED_EXIT
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/security/test_memory_provider_gate.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "scripts" / "security" / "memory_provider_gate.py"
+
+
+def metadata(**overrides):
+    data = {
+        "record_id": "rec-allowed",
+        "tenant_id": "tenant-a",
+        "user_id": "user-a",
+        "project_id": "project-a",
+        "workspace_id": "workspace-a",
+        "session_id": "session-a",
+        "branch": "main",
+        "baseline": "V1R4C3",
+        "agent_id": "agent-a",
+        "source_tool": "unit-test",
+        "created_at": "2026-07-10T00:00:00Z",
+        "schema_version": "1",
+        "sensitivity": "internal",
+        "retention_class": "ephemeral",
+    }
+    data.update(overrides)
+    return data
+
+
+def scope(**overrides):
+    data = {
+        "request_tenant_id": "tenant-a",
+        "request_user_id": "user-a",
+        "request_project_id": "project-a",
+        "request_workspace_id": "workspace-a",
+        "request_session_id": "session-a",
+        "current_branch": "main",
+        "current_baseline": "V1R4C3",
+        "request_agent_id": "agent-a",
+        "purpose": "unit-test",
+    }
+    data.update(overrides)
+    return data
+
+
+def run_gate(tmp_path: Path, candidate_records: list[dict], request_scope: dict | None = None):
+    scope_file = tmp_path / "scope.json"
+    candidates_file = tmp_path / "candidates.json"
+    output_file = tmp_path / "filtered.json"
+
+    scope_file.write_text(json.dumps(request_scope or scope()), encoding="utf-8")
+    candidates_file.write_text(json.dumps(candidate_records), encoding="utf-8")
+
+    proc = subprocess.run(
+        [
+            sys.executable,
+            str(SCRIPT),
+            "--scope-file",
+            str(scope_file),
+            "--candidates-file",
+            str(candidates_file),
+            "--output-file",
+            str(output_file),
+        ],
+        cwd=str(ROOT),
+        text=True,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        check=False,
+    )
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(output_file.read_text(encoding="utf-8"))
+
+
+def test_provider_gate_allows_only_scoped_candidates(tmp_path):
+    result = run_gate(
+        tmp_path,
+        [
+            {"metadata": metadata(), "body": "this body must not be required by the test"},
+            {"metadata": metadata(record_id="rec-denied", user_id="user-b"), "body": "redacted"},
+        ],
+    )
+
+    assert result["status"] == "filtered"
+    assert result["allowed_count"] == 1
+    assert result["denied_count"] == 1
+    assert result["allowed"][0]["record_id"] == "rec-allowed"
+    assert result["denied"][0]["record_id"] == "rec-denied"
+    assert "user_mismatch" in result["denied"][0]["reasons"]
+
+
+def test_provider_gate_never_forwards_denied_content(tmp_path):
+    result = run_gate(
+        tmp_path,
+        [
+            {
+                "metadata": metadata(record_id="rec-denied", tenant_id="tenant-b"),
+                "body": {"sensitive": "must-not-forward"},
+            }
+        ],
+    )
+
+    assert result["allowed"] == []
+    assert result["denied_count"] == 1
+    assert "body" not in result["denied"][0]
+    assert "tenant_mismatch" in result["denied"][0]["reasons"]
+
+
+def test_provider_gate_denies_untagged_candidate(tmp_path):
+    result = run_gate(tmp_path, [{"body": "untagged"}])
+
+    assert result["status"] == "filtered"
+    assert result["allowed_count"] == 0
+    assert result["denied_count"] == 1
+    assert any(reason.startswith("missing_metadata") for reason in result["denied"][0]["reasons"])
+
+
+def test_provider_gate_requires_current_branch_and_baseline(tmp_path):
+    result = run_gate(
+        tmp_path,
+        [
+            {"metadata": metadata(record_id="old-branch", branch="old")},
+            {"metadata": metadata(record_id="old-baseline", baseline="V1R1C0")},
+        ],
+    )
+
+    reasons = {item["record_id"]: item["reasons"] for item in result["denied"]}
+    assert "branch_mismatch" in reasons["old-branch"]
+    assert "baseline_mismatch" in reasons["old-baseline"]
+
+
+def test_provider_gate_can_allow_explicit_agent_allowlist(tmp_path):
+    result = run_gate(
+        tmp_path,
+        [{"metadata": metadata(record_id="agent-b", agent_id="agent-b")}],
+        request_scope=scope(allowed_agent_ids=["agent-a", "agent-b"]),
+    )
+
+    assert result["status"] == "healthy"
+    assert result["allowed_count"] == 1
+    assert result["denied_count"] == 0

tests/security/test_memory_scope_guard.py

@@ -0,0 +1,211 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT))
+
+from scripts.security.memory_scope_guard import (  # noqa: E402
+    record_matches_scope,
+    scan,
+    validate_scope_envelope,
+)
+
+
+def base_metadata(**overrides):
+    data = {
+        "record_id": "rec-1",
+        "tenant_id": "tenant-a",
+        "user_id": "user-a",
+        "project_id": "project-a",
+        "workspace_id": "workspace-a",
+        "session_id": "session-a",
+        "branch": "main",
+        "baseline": "V1R4C3",
+        "agent_id": "agent-a",
+        "source_tool": "unit-test",
+        "created_at": "2026-07-10T00:00:00Z",
+        "schema_version": "1",
+        "sensitivity": "internal",
+        "retention_class": "ephemeral",
+    }
+    data.update(overrides)
+    return data
+
+
+def base_scope(**overrides):
+    data = {
+        "request_tenant_id": "tenant-a",
+        "request_user_id": "user-a",
+        "request_project_id": "project-a",
+        "request_workspace_id": "workspace-a",
+        "request_session_id": "session-a",
+        "current_branch": "main",
+        "current_baseline": "V1R4C3",
+        "purpose": "unit-test",
+    }
+    data.update(overrides)
+    return data
+
+
+def scoped_path(root: Path, metadata: dict) -> Path:
+    return (
+        root
+        / "tenant"
+        / metadata["tenant_id"]
+        / "user"
+        / metadata["user_id"]
+        / "project"
+        / metadata["project_id"]
+        / "session"
+        / metadata["session_id"]
+        / f"{metadata['record_id']}.json"
+    )
+
+
+def write_json_record(root: Path, metadata: dict, body=None) -> Path:
+    path = scoped_path(root, metadata)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps({"metadata": metadata, "body": body or {"redacted": True}}, ensure_ascii=False),
+        encoding="utf-8",
+    )
+    return path
+
+
+def write_md_record(root: Path, metadata: dict) -> Path:
+    path = scoped_path(root, metadata).with_suffix(".md")
+    path.parent.mkdir(parents=True, exist_ok=True)
+    lines = ["---"]
+    for key, value in metadata.items():
+        lines.append(f"{key}: {value}")
+    lines.extend(["---", "", "redacted body"])
+    path.write_text("\n".join(lines), encoding="utf-8")
+    return path
+
+
+def test_same_user_project_session_record_is_allowed(tmp_path):
+    metadata = base_metadata()
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "healthy"
+    assert report["counts"]["allowed"] == 1
+    assert report["counts"]["denied"] == 0
+
+
+def test_markdown_frontmatter_record_is_allowed_when_scoped(tmp_path):
+    metadata = base_metadata(record_id="rec-md")
+    write_md_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "healthy"
+    assert report["counts"]["allowed"] == 1
+
+
+def test_different_user_is_denied(tmp_path):
+    metadata = base_metadata(user_id="user-b")
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert report["counts"]["denied"] == 1
+    assert "user_mismatch" in report["denied"][0]["reasons"]
+
+
+def test_different_project_is_denied(tmp_path):
+    metadata = base_metadata(project_id="project-b")
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert "project_mismatch" in report["denied"][0]["reasons"]
+
+
+def test_different_tenant_is_denied(tmp_path):
+    metadata = base_metadata(tenant_id="tenant-b")
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert "tenant_mismatch" in report["denied"][0]["reasons"]
+
+
+def test_untagged_record_is_denied(tmp_path):
+    path = tmp_path / "tenant" / "tenant-a" / "user" / "user-a" / "project" / "project-a" / "session" / "session-a" / "untagged.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps({"body": "redacted"}), encoding="utf-8")
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert any(reason.startswith("missing_metadata") for reason in report["denied"][0]["reasons"])
+
+
+def test_stale_branch_is_denied(tmp_path):
+    metadata = base_metadata(branch="old-feature-branch")
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert "branch_mismatch" in report["denied"][0]["reasons"]
+
+
+def test_stale_baseline_is_denied_as_current_evidence(tmp_path):
+    metadata = base_metadata(baseline="V1R1C0")
+    write_json_record(tmp_path, metadata)
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert "baseline_mismatch" in report["denied"][0]["reasons"]
+
+
+def test_unscoped_global_memory_path_is_denied(tmp_path):
+    metadata = base_metadata()
+    path = tmp_path / "global.json"
+    path.write_text(json.dumps({"metadata": metadata}), encoding="utf-8")
+
+    report = scan(tmp_path, base_scope(), require_scope=True)
+
+    assert report["status"] == "unhealthy"
+    assert "global_memory_path" in report["denied"][0]["reasons"]
+
+
+def test_scope_envelope_requires_core_dimensions():
+    missing = validate_scope_envelope({"request_user_id": "user-a"})
+
+    assert "request_project_id" in missing
+    assert "request_tenant_id_or_request_group_id" in missing
+
+
+def test_provider_fallback_without_scope_is_denied():
+    ok, reasons = record_matches_scope(base_metadata(), {"purpose": "fallback"})
+
+    assert ok is False
+    assert any(reason.startswith("invalid_scope") for reason in reasons)
+
+
+def test_cross_session_denied_by_default():
+    ok, reasons = record_matches_scope(base_metadata(session_id="session-b"), base_scope())
+
+    assert ok is False
+    assert "session_mismatch" in reasons
+
+
+def test_explicit_project_session_policy_can_allow_same_project_other_session():
+    ok, reasons = record_matches_scope(
+        base_metadata(session_id="session-b"),
+        base_scope(session_policy="allow_project_sessions"),
+    )
+
+    assert ok is True
+    assert reasons == []

tests/security/test_memory_search_activation_check.py

@@ -0,0 +1,116 @@
+from __future__ import annotations
+
+from scripts.security.memory_search_activation_check import evaluate
+
+
+def healthy_report(**overrides):
+    report = {
+        "status": "healthy",
+        "counts": {
+            "files": 3,
+            "allowed": 3,
+            "denied": 0,
+            "invalid": 0,
+        },
+    }
+    report.update(overrides)
+    return report
+
+
+def safe_provider(**overrides):
+    provider = {
+        "name": "safe-provider",
+        "scope_filter_before_ranking": True,
+        "provenance_required": True,
+        "fallback_mode": "text-only",
+        "allow_global_index": False,
+    }
+    provider.update(overrides)
+    return provider
+
+
+def test_activation_allowed_only_when_report_and_provider_are_safe():
+    result = evaluate(healthy_report(), safe_provider(), requested_enable=True)
+
+    assert result["status"] == "allowed"
+    assert result["reasons"] == []
+
+
+def test_activation_blocked_when_not_requested():
+    result = evaluate(healthy_report(), safe_provider(), requested_enable=False)
+
+    assert result["status"] == "blocked"
+    assert "memory_search_not_requested" in result["reasons"]
+
+
+def test_activation_blocked_when_isolation_report_unhealthy():
+    result = evaluate(healthy_report(status="unhealthy"), safe_provider(), requested_enable=True)
+
+    assert result["status"] == "blocked"
+    assert "isolation_report_not_healthy" in result["reasons"]
+
+
+def test_activation_blocked_when_denied_records_exist():
+    result = evaluate(
+        healthy_report(counts={"files": 2, "allowed": 1, "denied": 1, "invalid": 0}),
+        safe_provider(),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "denied_records_present" in result["reasons"]
+
+
+def test_activation_blocked_when_invalid_records_exist():
+    result = evaluate(
+        healthy_report(counts={"files": 2, "allowed": 1, "denied": 0, "invalid": 1}),
+        safe_provider(),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "invalid_records_present" in result["reasons"]
+
+
+def test_activation_blocked_without_filter_before_ranking():
+    result = evaluate(
+        healthy_report(),
+        safe_provider(scope_filter_before_ranking=False),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "provider_does_not_filter_before_ranking" in result["reasons"]
+
+
+def test_activation_blocked_without_provenance():
+    result = evaluate(
+        healthy_report(),
+        safe_provider(provenance_required=False),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "provider_does_not_return_provenance" in result["reasons"]
+
+
+def test_activation_blocked_for_unsafe_fallback():
+    result = evaluate(
+        healthy_report(),
+        safe_provider(fallback_mode="global-search"),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "provider_fallback_mode_unsafe" in result["reasons"]
+
+
+def test_activation_blocked_for_global_index():
+    result = evaluate(
+        healthy_report(),
+        safe_provider(allow_global_index=True),
+        requested_enable=True,
+    )
+
+    assert result["status"] == "blocked"
+    assert "global_index_not_allowed" in result["reasons"]