Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
7c38007094 feat(memory): add grounded observation synthesis layer
All checks were successful
Lint / lint (pull_request) Successful in 16s
2026-04-22 10:59:40 -04:00
6 changed files with 641 additions and 527 deletions

View File

@@ -1,70 +1,43 @@
from __future__ import annotations
"""
A2A agent card generation for fleet discovery.
Agent Card — A2A-compliant agent discovery.
Part of #843: fix: implement A2A agent card for fleet discovery (#819)
Refs #801.
Closes #802.
Provides metadata about the agent's identity, capabilities, and installed skills
for discovery by other agents in the fleet.
"""
import argparse
import json
import logging
import os
import socket
import sys
from dataclasses import asdict, dataclass, field
from typing import Any, Dict, Iterable, List, Mapping, Sequence
from urllib.parse import urlparse, urlunparse
from pathlib import Path
from typing import Any, Dict, List, Optional
from hermes_cli import __version__
from hermes_cli.config import load_config
from hermes_cli.config import load_config, get_hermes_home
from agent.skill_utils import (
get_all_skills_dirs,
get_disabled_skill_names,
iter_skill_index_files,
parse_frontmatter,
skill_matches_platform,
get_all_skills_dirs,
get_disabled_skill_names,
skill_matches_platform
)
logger = logging.getLogger(__name__)
DEFAULT_DESCRIPTION = "Sovereign AI agent — orchestration, code, research"
DEFAULT_INPUT_MODES = ["text/plain", "application/json"]
DEFAULT_OUTPUT_MODES = ["text/plain", "application/json"]
_REQUIRED_CAPABILITY_FLAGS = (
"streaming",
"pushNotifications",
"stateTransitionHistory",
)
@dataclass
class AgentSkill:
id: str
name: str
description: str = ""
tags: List[str] = field(default_factory=list)
def to_dict(self) -> Dict[str, Any]:
data: Dict[str, Any] = {"id": self.id, "name": self.name}
if self.description:
data["description"] = self.description
if self.tags:
data["tags"] = self.tags
return data
version: str = "1.0.0"
@dataclass
class AgentCapabilities:
streaming: bool = True
pushNotifications: bool = False
stateTransitionHistory: bool = True
def to_dict(self) -> Dict[str, Any]:
return asdict(self)
tools: bool = True
vision: bool = False
reasoning: bool = False
@dataclass
class AgentCard:
@@ -74,81 +47,14 @@ class AgentCard:
version: str = __version__
capabilities: AgentCapabilities = field(default_factory=AgentCapabilities)
skills: List[AgentSkill] = field(default_factory=list)
defaultInputModes: List[str] = field(default_factory=lambda: list(DEFAULT_INPUT_MODES))
defaultOutputModes: List[str] = field(default_factory=lambda: list(DEFAULT_OUTPUT_MODES))
metadata: Dict[str, Any] = field(default_factory=dict)
def to_dict(self) -> Dict[str, Any]:
data: Dict[str, Any] = {
"name": self.name,
"description": self.description,
"url": self.url,
"version": self.version,
"capabilities": self.capabilities.to_dict(),
"skills": [skill.to_dict() for skill in self.skills],
"defaultInputModes": list(self.defaultInputModes),
"defaultOutputModes": list(self.defaultOutputModes),
}
if self.metadata:
data["metadata"] = dict(self.metadata)
return data
def to_json(self, indent: int = 2) -> str:
return json.dumps(self.to_dict(), indent=indent)
def _env_or_empty(key: str) -> str:
return os.environ.get(key, "").strip()
def _as_agent_config(config: Mapping[str, Any] | None) -> Dict[str, Any]:
if not isinstance(config, Mapping):
return {}
agent_cfg = config.get("agent")
return dict(agent_cfg) if isinstance(agent_cfg, Mapping) else {}
def _as_a2a_config(config: Mapping[str, Any] | None) -> Dict[str, Any]:
if not isinstance(config, Mapping):
return {}
a2a_cfg = config.get("a2a")
return dict(a2a_cfg) if isinstance(a2a_cfg, Mapping) else {}
def _normalize_string_list(value: Any) -> List[str]:
if value is None:
return []
if isinstance(value, str):
parts = value.split(",")
elif isinstance(value, Sequence) and not isinstance(value, (bytes, bytearray, str)):
parts = list(value)
else:
parts = [value]
out: List[str] = []
seen = set()
for item in parts:
text = str(item).strip()
if not text or text in seen:
continue
seen.add(text)
out.append(text)
return out
def _normalize_skill_tags(frontmatter: Mapping[str, Any]) -> List[str]:
tags = _normalize_string_list(frontmatter.get("tags"))
category = str(frontmatter.get("category") or "").strip()
if category and category not in tags:
tags.append(category)
return tags
defaultInputModes: List[str] = field(default_factory=lambda: ["text/plain"])
defaultOutputModes: List[str] = field(default_factory=lambda: ["text/plain"])
def _load_skills() -> List[AgentSkill]:
"""Scan enabled skills and return A2A skill metadata."""
skills: List[AgentSkill] = []
"""Scan all enabled skills and return metadata."""
skills = []
disabled = get_disabled_skill_names()
seen_ids = set()
for skills_dir in get_all_skills_dirs():
if not skills_dir.is_dir():
continue
@@ -159,262 +65,71 @@ def _load_skills() -> List[AgentSkill]:
except Exception:
continue
skill_name = frontmatter.get("name") or skill_file.parent.name
if str(skill_name) in disabled:
continue
if not skill_matches_platform(frontmatter):
continue
skill_id = str(frontmatter.get("name") or skill_file.parent.name).strip().lower().replace(" ", "-")
if skill_id in disabled or skill_id in seen_ids:
continue
seen_ids.add(skill_id)
skills.append(AgentSkill(
id=str(skill_name),
name=str(frontmatter.get("name", skill_name)),
description=str(frontmatter.get("description", "")),
version=str(frontmatter.get("version", "1.0.0"))
))
return skills
display_name = str(frontmatter.get("title") or frontmatter.get("name") or skill_file.parent.name).strip()
description = str(frontmatter.get("description") or "").strip()
tags = _normalize_skill_tags(frontmatter)
skills.append(
AgentSkill(
id=skill_id,
name=display_name,
description=description,
tags=tags,
)
)
def build_agent_card() -> AgentCard:
"""Build the agent card from current configuration and environment."""
config = load_config()
# Identity
name = os.environ.get("HERMES_AGENT_NAME") or config.get("agent", {}).get("name") or "hermes"
description = os.environ.get("HERMES_AGENT_DESCRIPTION") or config.get("agent", {}).get("description") or "Sovereign AI agent"
# URL - try to determine from environment or config
port = os.environ.get("HERMES_WEB_PORT") or "9119"
host = os.environ.get("HERMES_WEB_HOST") or "localhost"
url = f"http://{host}:{port}"
# Capabilities
# In a real scenario, we'd check model metadata for vision/reasoning
capabilities = AgentCapabilities(
streaming=True,
tools=True,
vision=False, # Default to false unless we can confirm
reasoning=False
)
# Skills
skills = _load_skills()
return AgentCard(
name=name,
description=description,
url=url,
version=__version__,
capabilities=capabilities,
skills=skills
)
return sorted(skills, key=lambda skill: skill.id)
def _get_agent_name(config: Mapping[str, Any] | None, override: str | None = None) -> str:
if override:
return override
env_name = _env_or_empty("HERMES_AGENT_NAME") or _env_or_empty("AGENT_NAME")
if env_name:
return env_name
agent_cfg = _as_agent_config(config)
if agent_cfg.get("name"):
return str(agent_cfg["name"]).strip()
def get_agent_card_json() -> str:
"""Return the agent card as a JSON string."""
try:
hostname = socket.gethostname().split(".", 1)[0].strip()
if hostname:
return hostname
except Exception:
pass
return "hermes"
def _get_description(config: Mapping[str, Any] | None, override: str | None = None) -> str:
if override:
return override
env_description = _env_or_empty("HERMES_AGENT_DESCRIPTION") or _env_or_empty("AGENT_DESCRIPTION")
if env_description:
return env_description
agent_cfg = _as_agent_config(config)
if agent_cfg.get("description"):
return str(agent_cfg["description"]).strip()
return DEFAULT_DESCRIPTION
def _normalize_a2a_url(url: str) -> str:
raw = (url or "").strip()
if not raw:
return ""
parsed = urlparse(raw if "://" in raw else f"https://{raw}")
scheme = parsed.scheme or "https"
netloc = parsed.netloc or parsed.path
path = parsed.path if parsed.netloc else ""
normalized_path = path.rstrip("/") if path not in ("", "/") else ""
if not normalized_path.endswith("/a2a"):
normalized_path = f"{normalized_path}/a2a" if normalized_path else "/a2a"
return urlunparse((scheme, netloc, normalized_path, "", "", ""))
def _get_agent_url(config: Mapping[str, Any] | None, override: str | None = None) -> str:
if override:
return _normalize_a2a_url(override)
agent_cfg = _as_agent_config(config)
a2a_cfg = _as_a2a_config(config)
explicit = (
_env_or_empty("HERMES_A2A_PUBLIC_URL")
or str(a2a_cfg.get("public_url") or "").strip()
or str(agent_cfg.get("a2a_public_url") or "").strip()
)
if explicit:
return _normalize_a2a_url(explicit)
host = (
_env_or_empty("HERMES_A2A_HOST")
or str(a2a_cfg.get("host") or "").strip()
or _env_or_empty("HERMES_WEB_HOST")
or str(agent_cfg.get("host") or "").strip()
or "localhost"
)
port = (
_env_or_empty("HERMES_A2A_PORT")
or str(a2a_cfg.get("port") or "").strip()
or _env_or_empty("HERMES_WEB_PORT")
or str(agent_cfg.get("port") or "").strip()
or "9119"
)
scheme = (
_env_or_empty("HERMES_A2A_SCHEME")
or str(a2a_cfg.get("scheme") or "").strip()
or ("https" if (_env_or_empty("HERMES_MTLS_CERT") or str(port) == "9443") else "http")
)
return _normalize_a2a_url(f"{scheme}://{host}:{port}")
def _merge_skills(base_skills: Iterable[AgentSkill], extra_skills: Iterable[AgentSkill] | None = None) -> List[AgentSkill]:
merged: Dict[str, AgentSkill] = {}
for skill in list(base_skills) + list(extra_skills or []):
if skill.id not in merged:
merged[skill.id] = skill
return [merged[key] for key in sorted(merged)]
def build_agent_card(
*,
name: str | None = None,
description: str | None = None,
url: str | None = None,
extra_skills: Iterable[AgentSkill] | None = None,
metadata: Mapping[str, Any] | None = None,
) -> AgentCard:
"""Build an A2A-compliant agent card from config, env, and installed skills."""
try:
config = load_config()
except Exception as exc:
logger.debug("Falling back to empty config while building agent card: %s", exc)
config = {}
card = AgentCard(
name=_get_agent_name(config, override=name),
description=_get_description(config, override=description),
url=_get_agent_url(config, override=url),
skills=_merge_skills(_load_skills(), extra_skills),
metadata=dict(metadata or {}),
)
return card
def validate_agent_card(card: AgentCard | Dict[str, Any]) -> List[str]:
"""Return a list of schema-validation errors for an agent card."""
data = card.to_dict() if isinstance(card, AgentCard) else dict(card)
errors: List[str] = []
for field_name in ("name", "description", "url", "version"):
value = data.get(field_name)
if not isinstance(value, str) or not value.strip():
errors.append(f"{field_name} must be a non-empty string")
url_value = str(data.get("url") or "")
parsed = urlparse(url_value)
if not parsed.scheme or not parsed.netloc:
errors.append("url must be an absolute http/https URL")
elif parsed.scheme not in {"http", "https"}:
errors.append("url must use http or https")
elif not parsed.path.rstrip("/").endswith("/a2a"):
errors.append("url must point to the /a2a endpoint")
capabilities = data.get("capabilities")
if not isinstance(capabilities, Mapping):
errors.append("capabilities must be an object")
else:
for capability_name in _REQUIRED_CAPABILITY_FLAGS:
if not isinstance(capabilities.get(capability_name), bool):
errors.append(f"capabilities.{capability_name} must be a boolean")
for field_name, required_modes in (
("defaultInputModes", DEFAULT_INPUT_MODES),
("defaultOutputModes", DEFAULT_OUTPUT_MODES),
):
modes = data.get(field_name)
if not isinstance(modes, list) or not modes:
errors.append(f"{field_name} must be a non-empty list of MIME types")
continue
for mode in modes:
if not isinstance(mode, str) or "/" not in mode:
errors.append(f"{field_name} entries must be MIME types")
for required_mode in required_modes:
if required_mode not in modes:
errors.append(f"{field_name} must include {required_mode}")
skills = data.get("skills")
if not isinstance(skills, list):
errors.append("skills must be a list")
else:
for index, skill in enumerate(skills):
if not isinstance(skill, Mapping):
errors.append(f"skills[{index}] must be an object")
continue
if not str(skill.get("id") or "").strip():
errors.append(f"skills[{index}] missing id")
if not str(skill.get("name") or "").strip():
errors.append(f"skills[{index}] missing name")
tags = skill.get("tags", [])
if tags is None:
tags = []
if not isinstance(tags, list):
errors.append(f"skills[{index}].tags must be a list")
else:
for tag in tags:
if not isinstance(tag, str) or not tag.strip():
errors.append(f"skills[{index}].tags entries must be non-empty strings")
metadata = data.get("metadata")
if metadata is not None and not isinstance(metadata, Mapping):
errors.append("metadata must be an object when present")
return errors
def get_agent_card_json(
*,
name: str | None = None,
description: str | None = None,
url: str | None = None,
metadata: Mapping[str, Any] | None = None,
indent: int = 2,
) -> str:
"""Return the local agent card as JSON, falling back to an error card on failure."""
try:
card = build_agent_card(name=name, description=description, url=url, metadata=metadata)
errors = validate_agent_card(card)
if errors:
raise ValueError("; ".join(errors))
return card.to_json(indent=indent)
except Exception as exc:
logger.error("Failed to build agent card: %s", exc)
card = build_agent_card()
return json.dumps(asdict(card), indent=2)
except Exception as e:
logger.error(f"Failed to build agent card: {e}")
# Minimal fallback card
fallback = {
"name": name or _env_or_empty("HERMES_AGENT_NAME") or "hermes",
"description": "Sovereign AI agent (agent card fallback)",
"url": url or "http://localhost:9119/a2a",
"name": "hermes",
"description": "Sovereign AI agent (fallback)",
"version": __version__,
"capabilities": AgentCapabilities().to_dict(),
"skills": [],
"defaultInputModes": list(DEFAULT_INPUT_MODES),
"defaultOutputModes": list(DEFAULT_OUTPUT_MODES),
"error": str(exc),
"error": str(e)
}
return json.dumps(fallback, indent=indent)
return json.dumps(fallback, indent=2)
def main(argv: Sequence[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Generate an A2A-compliant Hermes agent card")
parser.add_argument("--name", help="Override the agent name")
parser.add_argument("--description", help="Override the agent description")
parser.add_argument("--url", help="Override the public A2A URL")
parser.add_argument("--validate", action="store_true", help="Validate before printing; exit 1 on schema errors")
args = parser.parse_args(list(argv) if argv is not None else None)
card = build_agent_card(name=args.name, description=args.description, url=args.url)
errors = validate_agent_card(card)
if args.validate and errors:
for error in errors:
print(error, file=sys.stderr)
return 1
print(card.to_json(indent=2))
return 0
if __name__ == "__main__":
raise SystemExit(main())
def validate_agent_card(card_data: Dict[str, Any]) -> bool:
"""Check if the card data complies with the A2A schema."""
required = ["name", "description", "url", "version"]
return all(k in card_data for k in required)

View File

@@ -26,6 +26,7 @@ from agent.memory_provider import MemoryProvider
from tools.registry import tool_error
from .store import MemoryStore
from .retrieval import FactRetriever
from .observations import ObservationSynthesizer
logger = logging.getLogger(__name__)
@@ -37,28 +38,29 @@ logger = logging.getLogger(__name__)
FACT_STORE_SCHEMA = {
"name": "fact_store",
"description": (
"Deep structured memory with algebraic reasoning. "
"Deep structured memory with algebraic reasoning and grounded observation synthesis. "
"Use alongside the memory tool — memory for always-on context, "
"fact_store for deep recall and compositional queries.\n\n"
"fact_store for deep recall, compositional queries, and higher-order observations.\n\n"
"ACTIONS (simple → powerful):\n"
"• add — Store a fact the user would expect you to remember.\n"
"• search — Keyword lookup ('editor config', 'deploy process').\n"
"• probe — Entity recall: ALL facts about a person/thing.\n"
"• related — What connects to an entity? Structural adjacency.\n"
"• reason — Compositional: facts connected to MULTIPLE entities simultaneously.\n"
"• observe — Synthesized higher-order observations backed by supporting facts.\n"
"• contradict — Memory hygiene: find facts making conflicting claims.\n"
"• update/remove/list — CRUD operations.\n\n"
"IMPORTANT: Before answering questions about the user, ALWAYS probe or reason first."
"IMPORTANT: Before answering questions about the user, ALWAYS probe/reason/observe first."
),
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["add", "search", "probe", "related", "reason", "contradict", "update", "remove", "list"],
"enum": ["add", "search", "probe", "related", "reason", "observe", "contradict", "update", "remove", "list"],
},
"content": {"type": "string", "description": "Fact content (required for 'add')."},
"query": {"type": "string", "description": "Search query (required for 'search')."},
"query": {"type": "string", "description": "Search query (required for 'search'/'observe')."},
"entity": {"type": "string", "description": "Entity name for 'probe'/'related'."},
"entities": {"type": "array", "items": {"type": "string"}, "description": "Entity names for 'reason'."},
"fact_id": {"type": "integer", "description": "Fact ID for 'update'/'remove'."},
@@ -66,6 +68,12 @@ FACT_STORE_SCHEMA = {
"tags": {"type": "string", "description": "Comma-separated tags."},
"trust_delta": {"type": "number", "description": "Trust adjustment for 'update'."},
"min_trust": {"type": "number", "description": "Minimum trust filter (default: 0.3)."},
"min_confidence": {"type": "number", "description": "Minimum observation confidence (default: 0.6)."},
"observation_type": {
"type": "string",
"enum": ["recurring_preference", "stable_direction", "behavioral_pattern"],
"description": "Optional observation type filter for 'observe'.",
},
"limit": {"type": "integer", "description": "Max results (default: 10)."},
},
"required": ["action"],
@@ -118,7 +126,9 @@ class HolographicMemoryProvider(MemoryProvider):
self._config = config or _load_plugin_config()
self._store = None
self._retriever = None
self._observation_synth = None
self._min_trust = float(self._config.get("min_trust_threshold", 0.3))
self._observation_min_confidence = float(self._config.get("observation_min_confidence", 0.6))
@property
def name(self) -> str:
@@ -177,6 +187,7 @@ class HolographicMemoryProvider(MemoryProvider):
hrr_weight=hrr_weight,
hrr_dim=hrr_dim,
)
self._observation_synth = ObservationSynthesizer(self._store)
self._session_id = session_id
def system_prompt_block(self) -> str:
@@ -193,30 +204,76 @@ class HolographicMemoryProvider(MemoryProvider):
"# Holographic Memory\n"
"Active. Empty fact store — proactively add facts the user would expect you to remember.\n"
"Use fact_store(action='add') to store durable structured facts about people, projects, preferences, decisions.\n"
"Use fact_store(action='observe') to synthesize higher-order observations with evidence.\n"
"Use fact_feedback to rate facts after using them (trains trust scores)."
)
return (
f"# Holographic Memory\n"
f"Active. {total} facts stored with entity resolution and trust scoring.\n"
f"Use fact_store to search, probe entities, reason across entities, or add facts.\n"
f"Use fact_store to search, probe entities, reason across entities, or synthesize observations.\n"
f"Use fact_feedback to rate facts after using them (trains trust scores)."
)
def prefetch(self, query: str, *, session_id: str = "") -> str:
if not self._retriever or not query:
if not query:
return ""
parts = []
raw_results = []
try:
results = self._retriever.search(query, min_trust=self._min_trust, limit=5)
if not results:
return ""
if self._retriever:
raw_results = self._retriever.search(query, min_trust=self._min_trust, limit=5)
except Exception as e:
logger.debug("Holographic prefetch fact search failed: %s", e)
raw_results = []
observations = []
try:
if self._observation_synth:
observations = self._observation_synth.observe(
query,
min_confidence=self._observation_min_confidence,
limit=3,
refresh=True,
)
except Exception as e:
logger.debug("Holographic prefetch observation search failed: %s", e)
observations = []
if not raw_results and observations:
seen_fact_ids = set()
evidence_backfill = []
for observation in observations:
for evidence in observation.get("evidence", []):
fact_id = evidence.get("fact_id")
if fact_id in seen_fact_ids:
continue
seen_fact_ids.add(fact_id)
evidence_backfill.append(evidence)
raw_results = evidence_backfill[:5]
if raw_results:
lines = []
for r in results:
for r in raw_results:
trust = r.get("trust_score", r.get("trust", 0))
lines.append(f"- [{trust:.1f}] {r.get('content', '')}")
return "## Holographic Memory\n" + "\n".join(lines)
except Exception as e:
logger.debug("Holographic prefetch failed: %s", e)
return ""
parts.append("## Holographic Memory\n" + "\n".join(lines))
if observations:
lines = []
for observation in observations:
evidence_ids = ", ".join(
f"#{item['fact_id']}" for item in observation.get("evidence", [])[:3]
) or "none"
lines.append(
f"- [{observation.get('confidence', 0.0):.2f}] "
f"{observation.get('observation_type', 'observation')}: "
f"{observation.get('summary', '')} "
f"(evidence: {evidence_ids})"
)
parts.append("## Holographic Observations\n" + "\n".join(lines))
return "\n\n".join(parts)
def sync_turn(self, user_content: str, assistant_content: str, *, session_id: str = "") -> None:
# Holographic memory stores explicit facts via tools, not auto-sync.
@@ -252,6 +309,7 @@ class HolographicMemoryProvider(MemoryProvider):
def shutdown(self) -> None:
self._store = None
self._retriever = None
self._observation_synth = None
# -- Tool handlers -------------------------------------------------------
@@ -305,6 +363,19 @@ class HolographicMemoryProvider(MemoryProvider):
)
return json.dumps({"results": results, "count": len(results)})
elif action == "observe":
synthesizer = self._observation_synth
if not synthesizer:
return tool_error("Observation synthesizer is not initialized")
observations = synthesizer.observe(
args.get("query", ""),
observation_type=args.get("observation_type"),
min_confidence=float(args.get("min_confidence", self._observation_min_confidence)),
limit=int(args.get("limit", 10)),
refresh=True,
)
return json.dumps({"observations": observations, "count": len(observations)})
elif action == "contradict":
results = retriever.contradict(
category=args.get("category"),

View File

@@ -0,0 +1,249 @@
"""Higher-order observation synthesis for holographic memory.
Builds grounded observations from accumulated facts and keeps them in a
separate retrieval layer with explicit evidence links back to supporting facts.
"""
from __future__ import annotations
import re
from typing import Any
from .store import MemoryStore
_TOKEN_RE = re.compile(r"[a-z0-9_]+")
_HIGHER_ORDER_CUES = {
"prefer",
"preference",
"preferences",
"style",
"pattern",
"patterns",
"behavior",
"behaviour",
"habit",
"habits",
"workflow",
"direction",
"trajectory",
"strategy",
"tend",
"usually",
}
_OBSERVATION_PATTERNS = [
{
"observation_type": "recurring_preference",
"subject": "communication_style",
"categories": {"user_pref", "general"},
"labels": {
"concise": ["concise", "terse", "brief", "short", "no fluff"],
"result_first": ["result-only", "result only", "outcome only", "quick", "quickly"],
"silent_ops": ["silent", "no status", "no repetitive status", "no questions"],
},
"summary_prefix": "Recurring preference",
},
{
"observation_type": "stable_direction",
"subject": "project_direction",
"categories": {"project", "general", "tool"},
"labels": {
"local_first": ["local-first", "local first", "local-only", "local only", "ollama", "own hardware"],
"gitea_first": ["gitea-first", "gitea first", "forge", "pull request", "pr flow", "issue flow"],
"ansible": ["ansible", "playbook", "role", "deploy via ansible"],
},
"summary_prefix": "Stable direction",
},
{
"observation_type": "behavioral_pattern",
"subject": "operator_workflow",
"categories": {"general", "project", "tool", "user_pref"},
"labels": {
"commit_early": ["commit early", "commits early", "commit after", "wip commit"],
"pr_first": ["open pr", "push a pr", "pull request", "pr immediately", "create pr"],
"dedup_guard": ["no dupes", "no duplicates", "avoid duplicate", "existing pr"],
},
"summary_prefix": "Behavioral pattern",
},
]
_TYPE_QUERY_HINTS = {
"recurring_preference": {"prefer", "preference", "style", "communication", "likes", "wants"},
"stable_direction": {"direction", "trajectory", "strategy", "project", "roadmap", "moving"},
"behavioral_pattern": {"pattern", "behavior", "workflow", "habit", "operator", "agent", "usually"},
}
class ObservationSynthesizer:
"""Synthesizes grounded observations from facts and retrieves them by query."""
def __init__(self, store: MemoryStore):
self.store = store
def synthesize(
self,
*,
persist: bool = True,
min_confidence: float = 0.6,
limit: int = 10,
) -> list[dict[str, Any]]:
facts = self.store.list_facts(min_trust=0.0, limit=1000)
observations: list[dict[str, Any]] = []
for pattern in _OBSERVATION_PATTERNS:
candidate = self._build_candidate(pattern, facts, min_confidence=min_confidence)
if not candidate:
continue
if persist:
candidate["observation_id"] = self.store.upsert_observation(
candidate["observation_type"],
candidate["subject"],
candidate["summary"],
candidate["confidence"],
candidate["evidence_fact_ids"],
metadata=candidate["metadata"],
)
candidate["evidence"] = self._expand_evidence(candidate["evidence_fact_ids"])
candidate["evidence_count"] = len(candidate["evidence"])
candidate.pop("evidence_fact_ids", None)
observations.append(candidate)
observations.sort(
key=lambda item: (item["confidence"], item.get("evidence_count", 0)),
reverse=True,
)
return observations[:limit]
def observe(
self,
query: str = "",
*,
observation_type: str | None = None,
min_confidence: float = 0.6,
limit: int = 10,
refresh: bool = True,
) -> list[dict[str, Any]]:
if refresh:
self.synthesize(persist=True, min_confidence=min_confidence, limit=limit)
observations = self.store.list_observations(
observation_type=observation_type,
min_confidence=min_confidence,
limit=max(limit * 4, 20),
)
if not observations:
return []
if not query:
return observations[:limit]
query_tokens = self._tokenize(query)
is_higher_order = bool(query_tokens & _HIGHER_ORDER_CUES)
ranked: list[dict[str, Any]] = []
for item in observations:
searchable = " ".join(
[
item.get("summary", ""),
item.get("subject", ""),
item.get("observation_type", ""),
" ".join(item.get("metadata", {}).get("labels", [])),
]
)
overlap = self._overlap_score(query_tokens, self._tokenize(searchable))
type_bonus = self._type_bonus(query_tokens, item.get("observation_type", ""))
if overlap <= 0 and type_bonus <= 0 and not is_higher_order:
continue
ranked_item = dict(item)
ranked_item["score"] = round(item.get("confidence", 0.0) + overlap + type_bonus, 3)
ranked.append(ranked_item)
if not ranked and is_higher_order:
ranked = [
{**item, "score": round(float(item.get("confidence", 0.0)), 3)}
for item in observations
]
ranked.sort(
key=lambda item: (item.get("score", 0.0), item.get("confidence", 0.0), item.get("evidence_count", 0)),
reverse=True,
)
return ranked[:limit]
def _build_candidate(
self,
pattern: dict[str, Any],
facts: list[dict[str, Any]],
*,
min_confidence: float,
) -> dict[str, Any] | None:
matched_fact_ids: set[int] = set()
matched_labels: dict[str, set[int]] = {label: set() for label in pattern["labels"]}
for fact in facts:
if fact.get("category") not in pattern["categories"]:
continue
haystack = f"{fact.get('content', '')} {fact.get('tags', '')}".lower()
local_match = False
for label, keywords in pattern["labels"].items():
if any(keyword in haystack for keyword in keywords):
matched_labels[label].add(int(fact["fact_id"]))
local_match = True
if local_match:
matched_fact_ids.add(int(fact["fact_id"]))
if len(matched_fact_ids) < 2:
return None
active_labels = sorted(label for label, ids in matched_labels.items() if ids)
confidence = min(0.95, 0.35 + 0.12 * len(matched_fact_ids) + 0.08 * len(active_labels))
confidence = round(confidence, 3)
if confidence < min_confidence:
return None
label_summary = ", ".join(label.replace("_", "-") for label in active_labels)
subject_text = pattern["subject"].replace("_", " ")
summary = (
f"{pattern['summary_prefix']}: {subject_text} trends toward {label_summary} "
f"based on {len(matched_fact_ids)} supporting facts."
)
return {
"observation_type": pattern["observation_type"],
"subject": pattern["subject"],
"summary": summary,
"confidence": confidence,
"metadata": {
"labels": active_labels,
"evidence_count": len(matched_fact_ids),
},
"evidence_fact_ids": sorted(matched_fact_ids),
}
def _expand_evidence(self, fact_ids: list[int]) -> list[dict[str, Any]]:
facts_by_id = {
fact["fact_id"]: fact
for fact in self.store.list_facts(min_trust=0.0, limit=1000)
}
return [facts_by_id[fact_id] for fact_id in fact_ids if fact_id in facts_by_id]
@staticmethod
def _tokenize(text: str) -> set[str]:
return set(_TOKEN_RE.findall(text.lower()))
@staticmethod
def _overlap_score(query_tokens: set[str], text_tokens: set[str]) -> float:
if not query_tokens or not text_tokens:
return 0.0
overlap = query_tokens & text_tokens
if not overlap:
return 0.0
return round(len(overlap) / max(len(query_tokens), 1), 3)
@staticmethod
def _type_bonus(query_tokens: set[str], observation_type: str) -> float:
hints = _TYPE_QUERY_HINTS.get(observation_type, set())
if not hints:
return 0.0
return 0.25 if query_tokens & hints else 0.0

View File

@@ -3,6 +3,7 @@ SQLite-backed fact store with entity resolution and trust scoring.
Single-user Hermes memory store plugin.
"""
import json
import re
import sqlite3
import threading
@@ -73,6 +74,28 @@ CREATE TABLE IF NOT EXISTS memory_banks (
fact_count INTEGER DEFAULT 0,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE IF NOT EXISTS observations (
observation_id INTEGER PRIMARY KEY AUTOINCREMENT,
observation_type TEXT NOT NULL,
subject TEXT NOT NULL,
summary TEXT NOT NULL,
confidence REAL DEFAULT 0.0,
metadata_json TEXT DEFAULT '{}',
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(observation_type, subject)
);
CREATE TABLE IF NOT EXISTS observation_evidence (
observation_id INTEGER REFERENCES observations(observation_id) ON DELETE CASCADE,
fact_id INTEGER REFERENCES facts(fact_id) ON DELETE CASCADE,
evidence_weight REAL DEFAULT 1.0,
PRIMARY KEY (observation_id, fact_id)
);
CREATE INDEX IF NOT EXISTS idx_observations_type ON observations(observation_type);
CREATE INDEX IF NOT EXISTS idx_observations_confidence ON observations(confidence DESC);
"""
# Trust adjustment constants
@@ -128,6 +151,7 @@ class MemoryStore:
def _init_db(self) -> None:
"""Create tables, indexes, and triggers if they do not exist. Enable WAL mode."""
self._conn.execute("PRAGMA journal_mode=WAL")
self._conn.execute("PRAGMA foreign_keys=ON")
self._conn.executescript(_SCHEMA)
# Migrate: add hrr_vector column if missing (safe for existing databases)
columns = {row[1] for row in self._conn.execute("PRAGMA table_info(facts)").fetchall()}
@@ -346,6 +370,115 @@ class MemoryStore:
rows = self._conn.execute(sql, params).fetchall()
return [self._row_to_dict(r) for r in rows]
def upsert_observation(
self,
observation_type: str,
subject: str,
summary: str,
confidence: float,
evidence_fact_ids: list[int],
metadata: dict | None = None,
) -> int:
"""Create or update a synthesized observation and its evidence links."""
with self._lock:
metadata_json = json.dumps(metadata or {}, sort_keys=True)
self._conn.execute(
"""
INSERT INTO observations (
observation_type, subject, summary, confidence, metadata_json
)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(observation_type, subject) DO UPDATE SET
summary = excluded.summary,
confidence = excluded.confidence,
metadata_json = excluded.metadata_json,
updated_at = CURRENT_TIMESTAMP
""",
(observation_type, subject, summary, confidence, metadata_json),
)
row = self._conn.execute(
"""
SELECT observation_id
FROM observations
WHERE observation_type = ? AND subject = ?
""",
(observation_type, subject),
).fetchone()
observation_id = int(row["observation_id"])
self._conn.execute(
"DELETE FROM observation_evidence WHERE observation_id = ?",
(observation_id,),
)
unique_fact_ids = sorted({int(fid) for fid in evidence_fact_ids})
if unique_fact_ids:
self._conn.executemany(
"""
INSERT OR IGNORE INTO observation_evidence (observation_id, fact_id)
VALUES (?, ?)
""",
[(observation_id, fact_id) for fact_id in unique_fact_ids],
)
self._conn.commit()
return observation_id
def list_observations(
self,
observation_type: str | None = None,
min_confidence: float = 0.0,
limit: int = 50,
) -> list[dict]:
"""List synthesized observations with expanded supporting evidence."""
with self._lock:
params: list = [min_confidence]
observation_clause = ""
if observation_type is not None:
observation_clause = "AND observation_type = ?"
params.append(observation_type)
params.append(limit)
rows = self._conn.execute(
f"""
SELECT observation_id, observation_type, subject, summary, confidence,
metadata_json, created_at, updated_at,
(
SELECT COUNT(*)
FROM observation_evidence oe
WHERE oe.observation_id = observations.observation_id
) AS evidence_count
FROM observations
WHERE confidence >= ?
{observation_clause}
ORDER BY confidence DESC, updated_at DESC
LIMIT ?
""",
params,
).fetchall()
results = []
for row in rows:
item = dict(row)
try:
item["metadata"] = json.loads(item.pop("metadata_json") or "{}")
except json.JSONDecodeError:
item["metadata"] = {}
item["evidence"] = self._get_observation_evidence(int(item["observation_id"]))
results.append(item)
return results
def _get_observation_evidence(self, observation_id: int) -> list[dict]:
rows = self._conn.execute(
"""
SELECT f.fact_id, f.content, f.category, f.tags, f.trust_score,
f.retrieval_count, f.helpful_count, f.created_at, f.updated_at
FROM observation_evidence oe
JOIN facts f ON f.fact_id = oe.fact_id
WHERE oe.observation_id = ?
ORDER BY f.trust_score DESC, f.updated_at DESC
""",
(observation_id,),
).fetchall()
return [self._row_to_dict(row) for row in rows]
def record_feedback(self, fact_id: int, helpful: bool) -> dict:
"""Record user feedback and adjust trust asymmetrically.

View File

@@ -1,150 +0,0 @@
from __future__ import annotations
import json
from pathlib import Path
import pytest
from agent import agent_card as mod
DEFAULT_DESCRIPTION = "Sovereign AI agent — orchestration, code, research"
def _set_base_context(monkeypatch, *, name: str = "Timmy", description: str = DEFAULT_DESCRIPTION, url: str = "https://timmy.local:9443/a2a", skills=None):
monkeypatch.setattr(mod, "load_config", lambda: {"agent": {"name": name, "description": description}})
monkeypatch.setattr(
mod,
"_load_skills",
lambda: list(
skills
if skills is not None
else [
mod.AgentSkill(
id="code",
name="Code Implementation",
description="Implement and patch code",
tags=["python", "gitea"],
)
]
),
)
monkeypatch.setenv("HERMES_A2A_PUBLIC_URL", url)
monkeypatch.delenv("HERMES_AGENT_NAME", raising=False)
monkeypatch.delenv("AGENT_NAME", raising=False)
monkeypatch.delenv("HERMES_AGENT_DESCRIPTION", raising=False)
monkeypatch.delenv("AGENT_DESCRIPTION", raising=False)
def test_build_agent_card_matches_issue_802_schema(monkeypatch):
_set_base_context(monkeypatch)
card = mod.build_agent_card()
payload = card.to_dict()
assert payload["name"] == "Timmy"
assert payload["description"] == DEFAULT_DESCRIPTION
assert payload["url"] == "https://timmy.local:9443/a2a"
assert payload["capabilities"] == {
"streaming": True,
"pushNotifications": False,
"stateTransitionHistory": True,
}
assert payload["defaultInputModes"] == ["text/plain", "application/json"]
assert payload["defaultOutputModes"] == ["text/plain", "application/json"]
assert payload["skills"][0]["tags"] == ["python", "gitea"]
assert mod.validate_agent_card(payload) == []
@pytest.mark.parametrize(
("name", "url"),
[
("Timmy", "https://timmy.local:9443/a2a"),
("Allegro", "https://allegro.local:9443/a2a"),
("Ezra", "https://ezra.local:9443/a2a"),
],
)
def test_build_agent_card_supports_fleet_members(monkeypatch, name, url):
_set_base_context(monkeypatch, name=name, url=url, skills=[])
payload = mod.build_agent_card().to_dict()
assert payload["name"] == name
assert payload["url"] == url
assert mod.validate_agent_card(payload) == []
def test_load_skills_collects_tags_and_category(monkeypatch, tmp_path):
skill_root = tmp_path / "skills"
skill_dir = skill_root / "code-implementation"
skill_dir.mkdir(parents=True)
(skill_dir / "SKILL.md").write_text(
"""---
name: Code Implementation
description: Implement and patch code
tags: [python, gitea]
category: discovery
---
# Code Implementation
""",
encoding="utf-8",
)
monkeypatch.setattr(mod, "get_all_skills_dirs", lambda: [skill_root])
monkeypatch.setattr(mod, "get_disabled_skill_names", lambda: set())
monkeypatch.setattr(mod, "skill_matches_platform", lambda _frontmatter: True)
skills = mod._load_skills()
assert len(skills) == 1
assert skills[0].id == "code-implementation"
assert skills[0].name == "Code Implementation"
assert skills[0].description == "Implement and patch code"
assert skills[0].tags == ["python", "gitea", "discovery"]
def test_validate_agent_card_reports_schema_errors():
errors = mod.validate_agent_card(
{
"name": "",
"description": "",
"url": "timmy.local",
"version": "",
"capabilities": {"streaming": True},
"skills": [{"id": "", "name": "", "tags": "python"}],
"defaultInputModes": ["text/plain"],
"defaultOutputModes": ["plain"],
"metadata": [],
}
)
assert any("name must be a non-empty string" in error for error in errors)
assert any("url must be an absolute http/https URL" in error for error in errors)
assert any("capabilities.pushNotifications" in error for error in errors)
assert any("skills[0] missing id" in error for error in errors)
assert any("skills[0].tags must be a list" in error for error in errors)
assert any("defaultInputModes must include application/json" in error for error in errors)
assert any("defaultOutputModes entries must be MIME types" in error for error in errors)
assert any("metadata must be an object" in error for error in errors)
def test_get_agent_card_json_emits_valid_json(monkeypatch):
_set_base_context(monkeypatch)
payload = json.loads(mod.get_agent_card_json())
assert payload["name"] == "Timmy"
assert mod.validate_agent_card(payload) == []
def test_main_validate_prints_card(monkeypatch, capsys):
_set_base_context(monkeypatch)
exit_code = mod.main(["--validate"])
captured = capsys.readouterr()
assert exit_code == 0
payload = json.loads(captured.out)
assert payload["url"] == "https://timmy.local:9443/a2a"
assert captured.err == ""

View File

@@ -0,0 +1,96 @@
import json
import pytest
from plugins.memory.holographic import HolographicMemoryProvider
from plugins.memory.holographic.store import MemoryStore
@pytest.fixture()
def store(tmp_path):
db_path = tmp_path / "memory.db"
s = MemoryStore(db_path=str(db_path), default_trust=0.5)
yield s
s.close()
@pytest.fixture()
def provider(tmp_path):
p = HolographicMemoryProvider(
config={
"db_path": str(tmp_path / "memory.db"),
"default_trust": 0.5,
}
)
p.initialize(session_id="test-session")
yield p
if p._store:
p._store.close()
class TestObservationSynthesis:
def test_observe_action_persists_observation_with_evidence_links(self, provider):
fact_ids = [
provider._store.add_fact('User prefers concise status updates', category='user_pref'),
provider._store.add_fact('User wants result-only replies with no fluff', category='user_pref'),
]
result = json.loads(
provider.handle_tool_call(
'fact_store',
{
'action': 'observe',
'query': 'What communication style does the user prefer?',
'limit': 5,
},
)
)
assert result['count'] == 1
observation = result['observations'][0]
assert observation['observation_type'] == 'recurring_preference'
assert observation['confidence'] >= 0.6
assert sorted(item['fact_id'] for item in observation['evidence']) == sorted(fact_ids)
stored = provider._store.list_observations(limit=10)
assert len(stored) == 1
assert stored[0]['observation_type'] == 'recurring_preference'
assert stored[0]['evidence_count'] == 2
assert len(provider._store.list_facts(limit=10)) == 2
def test_observe_action_synthesizes_three_observation_types(self, provider):
provider._store.add_fact('User prefers concise updates', category='user_pref')
provider._store.add_fact('User wants result-only communication', category='user_pref')
provider._store.add_fact('Project is moving to a local-first deployment model', category='project')
provider._store.add_fact('Project direction stays Gitea-first for issue and PR flow', category='project')
provider._store.add_fact('Operator always commits early before moving on', category='general')
provider._store.add_fact('Operator pushes a PR immediately after each meaningful fix', category='general')
result = json.loads(provider.handle_tool_call('fact_store', {'action': 'observe', 'limit': 10}))
types = {item['observation_type'] for item in result['observations']}
assert {'recurring_preference', 'stable_direction', 'behavioral_pattern'} <= types
def test_single_fact_does_not_create_overconfident_observation(self, provider):
provider._store.add_fact('User prefers concise updates', category='user_pref')
result = json.loads(
provider.handle_tool_call(
'fact_store',
{'action': 'observe', 'query': 'What does the user prefer?', 'limit': 5},
)
)
assert result['count'] == 0
assert provider._store.list_observations(limit=10) == []
def test_prefetch_surfaces_observations_as_separate_layer(self, provider):
provider._store.add_fact('User prefers concise updates', category='user_pref')
provider._store.add_fact('User wants result-only communication', category='user_pref')
prefetch = provider.prefetch('What communication style does the user prefer?')
assert '## Holographic Observations' in prefetch
assert '## Holographic Memory' in prefetch
assert 'recurring_preference' in prefetch
assert 'evidence' in prefetch.lower()