File size: 5,478 Bytes
5655a42 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """Hindsight plugin constants and pure config normalizers (no I/O, no origin imports)."""
from __future__ import annotations
import contextlib
import json
import logging
import re
from typing import Any, List
# Log under the plugin package's own logger name (loader-path independent).
logger = logging.getLogger(__name__.rpartition(".")[0])
_DEFAULT_API_URL = "https://api.hindsight.vectorize.io"
_DEFAULT_LOCAL_URL = "http://localhost:8888"
# Keep in sync with tools/lazy_deps.py ("memory.hindsight") and plugin.yaml.
_MIN_CLIENT_VERSION = "0.6.1"
_DEFAULT_TIMEOUT = 120 # seconds — cloud API can take 30-40s per request
_DEFAULT_IDLE_TIMEOUT = 300 # seconds — Hindsight embedded daemon default
# ``metadata.source`` on retained memories is OPT-IN (AGENTS.md forbids
# on-by-default attribution tags): ``retain_source`` / HINDSIGHT_RETAIN_SOURCE.
_DEFAULT_RETAIN_SOURCE = ""
# Hindsight brand mark (eye ringed by graph nodes) for the recall/retain indicators.
_HINDSIGHT_GLYPH = "👁️"
# Hindsight 0.5.0 added ``update_mode='append'``; older APIs would silently
# overwrite prior turns under a stable document_id, so they keep the per-process id.
# Mirrors hindsight-integrations/openclaw — Hindsight 0.5.0 added `update_mode='append'` semantics on retain
# (vectorize-io/hindsight#932).
_MIN_VERSION_FOR_UPDATE_MODE_APPEND = "0.5.0"
_VALID_BUDGETS = {"low", "mid", "high"}
_PROVIDER_DEFAULT_MODELS = {
"openai": "gpt-4o-mini",
"anthropic": "claude-haiku-4-5",
"gemini": "gemini-3.6-flash",
"groq": "openai/gpt-oss-120b",
"openrouter": "qwen/qwen3.5-9b",
"minimax": "MiniMax-M2.7",
"ollama": "gemma3:12b",
"lmstudio": "local-model",
"openai_compatible": "your-model-name",
}
# The embedded daemon speaks OpenAI wire format for these providers.
_OPENAI_WIRE_PROVIDERS = {"openai_compatible", "openrouter"}
_OBSERVATION_SCOPE_KEYWORDS = {"per_tag", "combined", "all_combinations"}
def _parse_int_setting(value: Any, default: int) -> int:
"""Parse an integer config/env value, falling back on invalid input."""
if value is None or value == "":
return default
try:
return int(value)
except (TypeError, ValueError):
logger.warning("Invalid integer Hindsight setting %r; using default %s", value, default)
return default
def _daemon_llm_provider(provider: str) -> str:
return "openai" if provider in _OPENAI_WIRE_PROVIDERS else provider
def _normalize_retain_tags(value: Any) -> List[str]:
"""Normalize tag config/tool values to a deduplicated list of strings."""
if value is None:
return []
raw_items = value if isinstance(value, list) else [value]
if isinstance(value, str):
text = value.strip()
parsed = None
if text.startswith("["):
with contextlib.suppress(Exception):
parsed = json.loads(text)
raw_items = parsed if isinstance(parsed, list) else text.split(",")
normalized: list[str] = []
for item in raw_items:
tag = str(item).strip()
if tag and tag not in normalized:
normalized.append(tag)
return normalized
def _normalize_observation_scopes(value: Any) -> Any:
"""Normalize observation_scopes to a keyword string, ``list[list[str]]`` (one inner
list per consolidation pass), or ``None`` (Hindsight's ``combined`` default).
Accepts a keyword, a JSON-encoded list, a flat tag list (one scope) or a list of
tag-lists; anything unrecognized -> ``None`` so we never send an invalid payload."""
if isinstance(value, str):
text = value.strip()
if text in _OBSERVATION_SCOPE_KEYWORDS:
return text
if text.startswith("["):
try:
return _normalize_observation_scopes(json.loads(text))
except Exception:
return None
return None
if not isinstance(value, (list, tuple)):
return None
if all(isinstance(entry, str) for entry in value): # flat tag list -> one scope
value = [value]
scopes = [
[str(tag).strip() for tag in entry if str(tag).strip()] if isinstance(entry, (list, tuple))
else [entry.strip()] if isinstance(entry, str) and entry.strip() else []
for entry in value
]
return [s for s in scopes if s] or None
def _sanitize_bank_segment(value: str) -> str:
"""URL/filesystem-safe bank_id placeholder: runs outside ``[A-Za-z0-9_-]`` (per
``str.isalnum``) become one dash; leading/trailing ``-``/``_`` are stripped."""
# \w == str.isalnum() + "_" for str patterns, so this matches the per-char rule.
return re.sub(r"[^\w-]+", "-", str(value)).strip("-_") if value else ""
def _resolve_bank_id_template(template: str, fallback: str, **placeholders: str) -> str:
"""Render a bank_id template ({profile}, {workspace}, {platform}, {user}, {session}),
sanitizing each placeholder; the ``-``/``_`` runs empty placeholders leave are
collapsed (``hermes-{user}`` -> ``hermes``). Empty/invalid template -> *fallback*."""
if not template:
return fallback
try:
rendered = template.format(**{k: _sanitize_bank_segment(v) for k, v in placeholders.items()})
except (KeyError, IndexError) as exc:
logger.warning("Invalid bank_id_template %r: %s — using fallback %r",
template, exc, fallback)
return fallback
return re.sub(r"([-_])\1+", r"\1", rendered).strip("-_") or fallback
|