From 33ed0aafc02fb6f1a678805ff90386a4eadc5b50 Mon Sep 17 00:00:00 2001 From: nicholascole Date: Mon, 13 Jul 2026 16:18:37 -0700 Subject: [PATCH] feat: OCG-backed long-term agent memory with human good/bad feedback links MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Port of the SDK side of agentspan-ai/agentspan#298. - OCGMemoryStore: synchronous MemoryStore HTTP adapter over the OCG BFF (search / save / delete / list, plus minting signed good/bad capability feedback URLs). Bearer-token auth held client-side. - Agent params: semantic_memory, memory_summary_model, feedback_sink. - AgentRuntime pre-run hook injects relevant agent/user-scoped memories into a COPY of the agent's instructions; post-run hook distills the conversation via an internal summarizer sub-agent (structured MemorySummary output, reuses the agent's model unless memory_summary_model is set) and saves it as conversation:. Both hooks are best-effort and recursion-guarded — memory never fails the primary run. - Feedback is human-only: the runtime hands a FeedbackEvent (distilled summary + signed good/bad URLs) to feedback_sink for out-of-band delivery; the URLs are never shown to the agent's LLM. The compiled path registers a spawn-safe {agent}_feedback_sink worker (FeedbackSinkEntry) mirroring run()'s delivery. - Serializer emits longTermMemory {ocgUrl, credential, agent, user, scope, maxResults, summaryModel} and feedbackSink {taskName} so the server-side compiler can inline memory steps on the deployed/webhook path. credential is a server-resolvable secret NAME, never the raw client token. - Example examples/agents/120_ocg_memory.py (renumbered from upstream 118, which is taken here) and unit tests. Co-Authored-By: Claude Fable 5 --- examples/agents/120_ocg_memory.py | 91 ++++++ src/conductor/ai/agents/__init__.py | 10 + src/conductor/ai/agents/agent.py | 10 + src/conductor/ai/agents/config_serializer.py | 44 ++- src/conductor/ai/agents/ocg_memory.py | 284 ++++++++++++++++++ .../ai/agents/runtime/_worker_entries.py | 50 +++ src/conductor/ai/agents/runtime/runtime.py | 211 ++++++++++++- tests/unit/ai/test_config_serializer.py | 63 ++++ tests/unit/ai/test_ocg_memory_store.py | 253 ++++++++++++++++ 9 files changed, 1014 insertions(+), 2 deletions(-) create mode 100644 examples/agents/120_ocg_memory.py create mode 100644 src/conductor/ai/agents/ocg_memory.py create mode 100644 tests/unit/ai/test_ocg_memory_store.py diff --git a/examples/agents/120_ocg_memory.py b/examples/agents/120_ocg_memory.py new file mode 100644 index 00000000..83bcb7db --- /dev/null +++ b/examples/agents/120_ocg_memory.py @@ -0,0 +1,91 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. + +"""120 — OCG-backed long-term memory with human good/bad feedback links. + +Enable memory on an agent and the runtime does two things automatically: + + - BEFORE a run: relevant past memories (scoped to this agent/user) are + retrieved from OCG and injected into the prompt — no tool call needed. + - AFTER a run: the conversation is summarized (Claude-style: durable facts, + not the raw transcript) by a small internal summarizer agent and saved back + to OCG as a memory. + +Feedback is HUMAN-only. Agents never vote. Instead, the runtime hands a +``FeedbackEvent`` — including signed *capability URLs* (good/bad) — to the +agent's ``feedback_sink``. A human (e.g. a support engineer) clicks a link to +mark the memory good or bad; the link skips auth (its signature is the +authorization), so the clicker needs no OCG account. Here the sink just prints +the URLs as they'd appear in a Zendesk ticket comment. + +Requires the OCG instance to be started with a feedback-link secret +(``OCG_FEEDBACK_LINK_SECRET``) for the capability URLs to be minted. + +Run (from the repo root):: + + OCG_INSTANCE_URL=https://test.contextgraph.io \ + OCG_TOKEN= \ + uv run python examples/agents/120_ocg_memory.py + + # against an embedded server, also set AGENTSPAN_SERVER_URL. +""" + +import os + +from conductor.ai.agents import Agent, AgentRuntime, OCGMemoryStore, SemanticMemory +from conductor.ai.agents.ocg_memory import FeedbackEvent + +MODEL = os.environ.get("AGENTSPAN_LLM_MODEL", "openai/gpt-4o-mini") + +OCG_INSTANCE_URL = os.environ.get("OCG_INSTANCE_URL") or "" +# Unlike the ocg.py retrieval tools (which resolve a credential server-side), +# the memory store calls OCG directly from Python, so it holds the bearer token. +OCG_TOKEN = os.environ.get("OCG_TOKEN") +if not OCG_INSTANCE_URL: + raise SystemExit("Set OCG_INSTANCE_URL to your OCG instance, e.g. https://test.contextgraph.io") + + +def zendesk_sink(event: FeedbackEvent) -> None: + """Deliver the good/bad links to a human. In production this would POST a + comment to the Zendesk ticket; here we just print what would be sent.""" + print("\n--- would post to Zendesk ticket ---") + print(f"Saved memory: {event.memory_key}") + print(f"Summary: {event.summary}") + if event.good_url: + print(f" 👍 Was this helpful? {event.good_url}") + print(f" 👎 Not helpful: {event.bad_url}") + print("------------------------------------\n") + + +def main() -> None: + store = OCGMemoryStore( + url=OCG_INSTANCE_URL, + agent="agent:support", + user="user:alice", + token=OCG_TOKEN, + ) + + agent = Agent( + name="support", + model=MODEL, + instructions=( + "You are a customer support agent. Use any relevant context from " + "memory to personalize your answer. A memory labeled [bad] was " + "flagged by a human — treat it with suspicion." + ), + semantic_memory=SemanticMemory(store=store, max_results=5), + feedback_sink=zendesk_sink, + ) + + with AgentRuntime() as runtime: + print("--- Turn 1 ---") + runtime.run( + agent, "Hi, I'm Alice. I'm on the Enterprise plan and prefer email." + ).print_result() + + print("\n--- Turn 2 (should recall Alice's plan from memory) ---") + runtime.run(agent, "What plan am I on again?").print_result() + + +if __name__ == "__main__": + main() diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index 6bdeda04..3c45e4e9 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -193,6 +193,12 @@ def resolve_credentials(task: object, names: list) -> dict: schedules, ) from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore, SemanticMemory +from conductor.ai.agents.ocg_memory import ( + FeedbackEvent, + MemorySummary, + OCGMemoryStore, + build_memory_summarizer, +) # Termination conditions from conductor.ai.agents.termination import ( @@ -326,6 +332,10 @@ def resolve_credentials(task: object, names: list) -> dict: "SemanticMemory", "MemoryStore", "MemoryEntry", + "OCGMemoryStore", + "MemorySummary", + "FeedbackEvent", + "build_memory_summarizer", # Code execution "CodeExecutionConfig", "CliConfig", diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index f79f9dca..6b886ccf 100644 --- a/src/conductor/ai/agents/agent.py +++ b/src/conductor/ai/agents/agent.py @@ -562,6 +562,9 @@ def __init__( output_type: Optional[type] = None, guardrails: Optional[List[Any]] = None, memory: Optional[Any] = None, + semantic_memory: Optional[Any] = None, + memory_summary_model: Optional[str] = None, + feedback_sink: Optional[Callable[..., Any]] = None, dependencies: Optional[Dict[str, Any]] = None, max_turns: int = 25, max_tokens: Optional[int] = None, @@ -724,6 +727,13 @@ def __init__( self.output_type = output_type self.guardrails: List[Any] = list(guardrails) if guardrails else [] self.memory = memory + # OCG-backed long-term memory (see agents/ocg_memory.py). When set, the + # runtime auto-injects relevant memories into the prompt before a run and, + # after the run, summarizes the conversation into a memory. feedback_sink, + # if provided, receives the good/bad capability links for that memory. + self.semantic_memory = semantic_memory + self.memory_summary_model = memory_summary_model + self.feedback_sink = feedback_sink self.dependencies: Dict[str, Any] = dict(dependencies) if dependencies else {} self.max_turns = max_turns self.max_tokens = max_tokens diff --git a/src/conductor/ai/agents/config_serializer.py b/src/conductor/ai/agents/config_serializer.py index ee68c0ba..d153e531 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -130,10 +130,21 @@ def _serialize_agent(self, agent: "Agent") -> dict: if agent.guardrails: config["guardrails"] = [self._serialize_guardrail(g) for g in agent.guardrails] - # Memory + # Memory (short-term conversation) if hasattr(agent, "memory") and agent.memory: config["memory"] = self._serialize_memory(agent.memory) + # Long-term (OCG-backed) memory. When present, the server-side compiler + # inlines retrieval (pre-loop) + distill/save/feedback (post-loop) steps + # so memory works on the deployed/webhook path — not just client run(). + ltm = self._serialize_long_term_memory(agent) + if ltm is not None: + config["longTermMemory"] = ltm + # feedback_sink delivers the human good/bad capability links out-of-band. + # Emit a worker ref so the compiled path can call the Python sink worker. + if getattr(agent, "feedback_sink", None) is not None: + config["feedbackSink"] = {"taskName": f"{agent.name}_feedback_sink"} + # Max tokens if agent.max_tokens is not None: config["maxTokens"] = agent.max_tokens @@ -508,3 +519,34 @@ def _serialize_memory(self, memory: Any) -> dict: if hasattr(memory, "max_messages") and memory.max_messages: result["maxMessages"] = memory.max_messages return result + + def _serialize_long_term_memory(self, agent: "Agent") -> "Any": + """Serialize an agent's OCG-backed semantic memory to a LongTermMemoryConfig dict. + + Returns ``None`` (no-op) unless the agent has a ``semantic_memory`` whose + store exposes an OCG base url. Reads the OCG instance url, scope owner, + user and scope off the store; the credential is a SERVER-resolvable secret + NAME (e.g. ``OCG_PUBLIC_KEY``) — never the raw client token. The summary + model falls back to the agent's own model when not explicitly set. + """ + sm = getattr(agent, "semantic_memory", None) + if sm is None: + return None + store = getattr(sm, "store", None) + # Only OCG-backed stores compile server-side (need a base url to call). + base = getattr(store, "_base", None) if store is not None else None + if not base: + return None + + result: Dict[str, Any] = { + "ocgUrl": base, + "credential": getattr(store, "_credential", None) or "OCG_PUBLIC_KEY", + "agent": getattr(store, "_agent", None), + "scope": getattr(store, "_scope", None) or "agent", + "maxResults": getattr(sm, "max_results", None), + "summaryModel": getattr(agent, "memory_summary_model", None) or (agent.model or None), + } + user = getattr(store, "_user", None) + if user: + result["user"] = user + return {k: v for k, v in result.items() if v is not None} diff --git a/src/conductor/ai/agents/ocg_memory.py b/src/conductor/ai/agents/ocg_memory.py new file mode 100644 index 00000000..69f5500b --- /dev/null +++ b/src/conductor/ai/agents/ocg_memory.py @@ -0,0 +1,284 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. See LICENSE file in the project root for details. + +"""OCG-backed long-term memory for agents. + +This module backs agentspan's :class:`~conductor.ai.agents.semantic_memory.MemoryStore` +abstraction with an OCG (Open Context Graph) instance, so an agent's memories +persist in OCG and ride OCG's feedback-aware ranking. + +Three pieces: + +- :class:`OCGMemoryStore` — a synchronous HTTP adapter implementing ``MemoryStore`` + (``add`` / ``search`` / ``delete`` / ``clear`` / ``list_all``) against the OCG BFF. +- :class:`MemorySummary` + :func:`build_memory_summarizer` — a small agent that + distills a conversation into durable facts (used by the runtime's post-run save). +- :class:`FeedbackEvent` — what the runtime hands to an Agent's ``feedback_sink`` + after saving a memory: the distilled summary plus signed *capability URLs* a human + can click to mark the memory good/bad (no OCG account needed). + +Design notes: + +- The OCG bearer ``token`` is held **client-side** here (e.g. from ``OCG_TOKEN``), + unlike the ``ocg.py`` retrieval tools which resolve a credential server-side. +- Agents only ever **create and read** memories. Good/bad feedback is human-only: + it is delivered out-of-band through ``feedback_sink`` (e.g. into a Zendesk ticket) + and the capability URLs are never surfaced to the agent's LLM. +""" + +from __future__ import annotations + +import hashlib +import logging +from dataclasses import dataclass, field +from typing import TYPE_CHECKING, Any, Dict, List, Optional + +import httpx + +from conductor.ai.agents.exceptions import AgentAPIError +from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore + +if TYPE_CHECKING: + from conductor.ai.agents.agent import Agent + +logger = logging.getLogger("conductor.ai.agents.ocg_memory") + + +def _hash_key(content: str) -> str: + return "mem-" + hashlib.sha256(content.encode("utf-8")).hexdigest()[:16] + + +class OCGMemoryStore(MemoryStore): + """Back agentspan :class:`SemanticMemory` with an OCG instance. + + Implements the synchronous ``MemoryStore`` interface over the OCG BFF: + + - ``add`` -> ``POST /api/v1/memories`` + - ``search`` -> ``POST /api/v1/memories/search`` (feedback-blended ranking) + - ``delete`` -> ``DELETE /api/v1/memories/{key}`` + - ``list_all``-> ``GET /api/v1/memories`` + + Args: + url: Base URL of the OCG instance (required). + agent: Agent owner key, e.g. ``"agent:support"`` (required). + user: Optional user owner, e.g. ``"user:alice"``. + token: OCG bearer token, held client-side (e.g. from ``OCG_TOKEN``). + Used by the client-side ``run()`` path. + credential: Server-resolvable credential NAME (default ``"OCG_PUBLIC_KEY"``) + for the OCG bearer token. Used by the COMPILED/deployed path — the + server resolves this via a ``#{NAME}`` HTTP-header placeholder. Distinct + from ``token`` (the raw client token); both can coexist. + scope: Memory scope for writes (default ``"user"``). + timeout: Per-request timeout in seconds. + client: Optional pre-built ``httpx.Client`` (mainly for tests). + """ + + def __init__( + self, + *, + url: str, + agent: str, + user: Optional[str] = None, + token: Optional[str] = None, + credential: str = "OCG_PUBLIC_KEY", + scope: str = "user", + timeout: float = 10.0, + client: Optional[httpx.Client] = None, + ) -> None: + if not url or not url.strip(): + raise ValueError("OCGMemoryStore requires a non-blank OCG instance url") + if not agent or not agent.strip(): + raise ValueError("OCGMemoryStore requires a non-blank agent owner") + self._base = url.strip().rstrip("/") + self._agent = agent + self._user = user + self._credential = credential + self._scope = scope + headers: Dict[str, str] = {} + if token: + headers["Authorization"] = f"Bearer {token}" + self._client = client or httpx.Client(timeout=timeout, headers=headers) + + # ── HTTP plumbing ─────────────────────────────────────────────────── + + def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response: + try: + resp = self._client.request(method, self._base + path, **kwargs) + except httpx.HTTPError as exc: # network/timeout + raise AgentAPIError(status_code=0, message=str(exc), url=self._base + path) from exc + if resp.status_code >= 400: + raise AgentAPIError( + status_code=resp.status_code, + message=resp.text, + url=self._base + path, + ) + return resp + + # ── MemoryStore interface ─────────────────────────────────────────── + + def add(self, entry: MemoryEntry) -> str: + key = entry.id or str(entry.metadata.get("key") or "") or _hash_key(entry.content) + body: Dict[str, Any] = { + "key": key, + "agent": self._agent, + "value": entry.content, + "description": entry.content[:200], + "scope": self._scope, + "source": "agent_inferred", + "tags": list(entry.metadata.get("tags", []) or []), + } + if self._user: + body["user"] = self._user + self._request("POST", "/api/v1/memories", json=body) + entry.id = key + return key + + def search(self, query: str, top_k: int = 5) -> List[MemoryEntry]: + body: Dict[str, Any] = { + "query": query, + "agent": self._agent, + "limit": top_k, + "include_shared": True, + } + if self._user: + body["user"] = self._user + resp = self._request("POST", "/api/v1/memories/search", json=body) + out: List[MemoryEntry] = [] + for m in resp.json().get("memories", []) or []: + out.append( + MemoryEntry( + id=m.get("key", ""), + content=_with_signal(m.get("value_preview", ""), m), + metadata={ + "relevance_score": m.get("relevance_score"), + "good_count": m.get("good_count", 0), + "bad_count": m.get("bad_count", 0), + }, + ) + ) + return out + + def delete(self, memory_id: str) -> bool: + params: Dict[str, str] = {"agent": self._agent} + if self._user: + params["user"] = self._user + try: + self._request("DELETE", f"/api/v1/memories/{memory_id}", params=params) + except AgentAPIError: + return False + return True + + def clear(self) -> None: + # No bulk-clear endpoint — fan out over the listed keys. Guard usage: + # this deletes every memory for the configured agent/user. + entries = self.list_all() + logger.warning( + "OCGMemoryStore.clear() deleting %d memories for %s", len(entries), self._agent + ) + for e in entries: + self.delete(e.id) + + def list_all(self) -> List[MemoryEntry]: + params: Dict[str, str] = {"agent": self._agent, "limit": "200"} + if self._user: + params["user"] = self._user + resp = self._request("GET", "/api/v1/memories", params=params) + return [ + MemoryEntry(id=m.get("key", ""), content=m.get("value_preview", "")) + for m in resp.json().get("memories", []) or [] + ] + + # ── Capability feedback links (human-only, out-of-band) ───────────── + + def feedback_links(self, key: str) -> Dict[str, Any]: + """Mint signed good/bad capability URLs for a memory. + + Returns ``{"good_url", "bad_url", "expires_at"}``. The URLs require no OCG + login — a human (e.g. a support engineer) clicks them to vote. Requires the + OCG instance to have a feedback-link secret configured (else OCG returns 501). + """ + params: Dict[str, str] = {"agent": self._agent} + if self._user: + params["user"] = self._user + resp = self._request("POST", f"/api/v1/memories/{key}/feedback-links", params=params) + return resp.json() + + +def _with_signal(content: str, m: Dict[str, Any]) -> str: + """Fold the human good/bad signal into a search result's content so the + injected prompt context shows the agent when a memory was marked bad and why.""" + good = int(m.get("good_count", 0) or 0) + bad = int(m.get("bad_count", 0) or 0) + if not good and not bad: + return content + content += f" [good {good} / bad {bad}]" + for note in m.get("feedback_notes") or []: + if note.get("verdict") == "bad" and note.get("reason"): + content += f' (bad: "{note["reason"]}")' + return content + + +# ── Conversation summarization (Claude-style distillation) ────────────── + +try: # pydantic is a core dep, but keep the import resilient. + from pydantic import BaseModel, Field + + class MemorySummary(BaseModel): + """Structured output for the conversation summarizer agent.""" + + summary: str = Field(description="One short paragraph: what happened / what was learned.") + facts: List[str] = Field( + default_factory=list, + description="Durable, reusable facts about the user or task (no chit-chat).", + ) + tags: List[str] = Field(default_factory=list, description="Short topical tags.") + +except Exception: # pragma: no cover - pydantic always present in practice + MemorySummary = None # type: ignore[assignment] + + +MEMORY_SUMMARIZER_INSTRUCTIONS = ( + "You distill a conversation into a durable memory. Read the transcript and " + "extract only reusable, durable facts about the user, their preferences, and " + "the task — the kind of thing worth remembering for next time. Ignore greetings, " + "filler, and one-off details. Write a one-paragraph summary, a short list of " + "facts, and a few topical tags. Be concise and concrete." +) + + +def build_memory_summarizer(model: str, *, name: str = "__memory_summarizer") -> "Agent": + """Build the internal agent that summarizes a conversation into a memory. + + It uses :class:`MemorySummary` structured output and is intentionally created + WITHOUT ``semantic_memory`` so the post-run save hook skips it (no recursion). + """ + from conductor.ai.agents.agent import Agent + + return Agent( + name=name, + model=model, + instructions=MEMORY_SUMMARIZER_INSTRUCTIONS, + output_type=MemorySummary, + max_turns=1, + ) + + +@dataclass +class FeedbackEvent: + """Handed to an Agent's ``feedback_sink`` after a conversation memory is saved. + + Carries the distilled summary plus the signed capability URLs a human can click + to mark the memory good/bad. The integrator routes these out-of-band (e.g. posts + them into a Zendesk ticket). These URLs are never shown to the agent's LLM. + """ + + memory_key: str + summary: str + facts: List[str] = field(default_factory=list) + tags: List[str] = field(default_factory=list) + good_url: Optional[str] = None + bad_url: Optional[str] = None + expires_at: Optional[str] = None + agent: Optional[str] = None + user: Optional[str] = None + session_id: Optional[str] = None diff --git a/src/conductor/ai/agents/runtime/_worker_entries.py b/src/conductor/ai/agents/runtime/_worker_entries.py index 9373f13f..5793d623 100644 --- a/src/conductor/ai/agents/runtime/_worker_entries.py +++ b/src/conductor/ai/agents/runtime/_worker_entries.py @@ -431,6 +431,56 @@ async def __call__(self, result: str = "") -> object: return {"decision": "continue"} # safe fallback +class FeedbackSinkEntry: + """Long-term-memory feedback_sink worker. + + The compiled (server-side) memory path invokes this with the FeedbackEvent + fields after saving a conversation memory; it rebuilds the + :class:`~conductor.ai.agents.ocg_memory.FeedbackEvent` and hands it to the + user's ``feedback_sink`` callable. Best-effort: failures are swallowed so + memory delivery never fails the run. + """ + + def __init__(self, feedback_sink_fn): + self.fn_t = wrap_callable(feedback_sink_fn) + + async def __call__( + self, + memory_key: str = "", + summary: str = "", + facts: object = None, + tags: object = None, + good_url: str = None, + bad_url: str = None, + expires_at: str = None, + agent: str = None, + user: str = None, + ) -> object: + from conductor.ai.agents.runtime.runtime import _call_user_fn + + try: + from conductor.ai.agents.ocg_memory import FeedbackEvent + + event = FeedbackEvent( + memory_key=memory_key, + summary=summary, + facts=list(facts) if isinstance(facts, (list, tuple)) else [], + tags=list(tags) if isinstance(tags, (list, tuple)) else [], + good_url=good_url, + bad_url=bad_url, + expires_at=expires_at, + agent=agent, + user=user, + ) + await _call_user_fn(unwrap_callable(self.fn_t), event) + return {"delivered": True} + except Exception as e: + import logging + + logging.getLogger(__name__).warning("feedback_sink delivery failed: %s", e) + return {"delivered": False} + + class CallbackEntry: """Position callback worker (was ``callback_worker``). diff --git a/src/conductor/ai/agents/runtime/runtime.py b/src/conductor/ai/agents/runtime/runtime.py index 0c54363e..975eb21b 100644 --- a/src/conductor/ai/agents/runtime/runtime.py +++ b/src/conductor/ai/agents/runtime/runtime.py @@ -162,6 +162,38 @@ def _has_stateful_tools(agent: Any) -> bool: return False +def _agent_model_str(agent: Any) -> str: + """Best-effort string model id for the conversation summarizer.""" + m = getattr(agent, "model", "") + if isinstance(m, str) and m: + return m + return "openai/gpt-4o-mini" + + +def _parse_summary_output(output: Any) -> "tuple[str, list, list]": + """Extract (summary, facts, tags) from a summarizer run's output. + + Handles a pydantic MemorySummary, a plain dict (optionally wrapped under a + ``result`` key), or any other value (stringified as the summary). + """ + # pydantic model instance + if hasattr(output, "summary"): + return ( + str(getattr(output, "summary", "") or ""), + list(getattr(output, "facts", []) or []), + list(getattr(output, "tags", []) or []), + ) + if isinstance(output, dict): + data = output.get("result", output) + if isinstance(data, dict): + return ( + str(data.get("summary", "") or ""), + list(data.get("facts", []) or []), + list(data.get("tags", []) or []), + ) + return (str(output) if output is not None else "", [], []) + + # Thread count for system-level async workers (guardrails, handoff checks, etc.). # User-defined tool workers keep the per-worker default from @worker_task. _SYSTEM_WORKER_THREADS = 10 @@ -1001,6 +1033,13 @@ def _collect_worker_names(self, agent: Agent, *, required_workers: Optional[set] if agent.termination: names.add(f"{agent.name}_termination") + # Long-term (OCG) memory feedback_sink — compiled path emits a SIMPLE + # task that delivers the human good/bad capability links out-of-band. + if getattr(agent, "semantic_memory", None) is not None and callable( + getattr(agent, "feedback_sink", None) + ): + names.add(f"{agent.name}_feedback_sink") + # Callable gate (sequential pipeline) if getattr(agent, "gate", None) is not None and callable(agent.gate): names.add(f"{agent.name}_gate") @@ -1156,6 +1195,16 @@ def _server_needs(task_name: str) -> bool: if _server_needs(task_name): self._register_stop_when_worker(agent.name, agent.stop_when, domain=domain) + # 3a. Long-term memory feedback_sink — compiled path hands the human + # good/bad capability links to this worker after a conversation memory + # is saved (mirrors run()'s post-run FeedbackEvent delivery). + if getattr(agent, "semantic_memory", None) is not None and callable( + getattr(agent, "feedback_sink", None) + ): + task_name = f"{agent.name}_feedback_sink" + if _server_needs(task_name): + self._register_feedback_sink_worker(agent.name, agent.feedback_sink, domain=domain) + # 3b. Callbacks (legacy + CallbackHandler chaining) from conductor.ai.agents.callback import ( _LEGACY_ATTR_TO_POSITION, @@ -1428,6 +1477,51 @@ def _register_stop_when_worker( lease_extend_enabled=True, )(stop_when_worker) + def _register_feedback_sink_worker( + self, agent_name: str, feedback_sink_fn, domain: "Optional[str]" = None + ) -> None: + """Register a long-term-memory feedback_sink worker. + + The compiled (server-side) memory path emits a SIMPLE task that, after + saving a conversation memory and minting the signed good/bad capability + URLs, invokes this worker with the FeedbackEvent fields. The worker + rebuilds a :class:`FeedbackEvent` and hands it to the user's + ``feedback_sink`` callable — mirroring run()'s out-of-band delivery. + Best-effort: failures are swallowed so memory never fails the run. + """ + from conductor.client.worker.worker_task import worker_task + + task_name = f"{agent_name}_feedback_sink" + + from conductor.ai.agents.runtime._worker_entries import ( + FeedbackSinkEntry, + probe_spawn_safety, + ) + + feedback_sink_worker = FeedbackSinkEntry(feedback_sink_fn) + feedback_sink_worker.__annotations__ = { + "memory_key": str, + "summary": str, + "facts": object, + "tags": object, + "good_url": str, + "bad_url": str, + "expires_at": str, + "agent": str, + "user": str, + "return": object, + } + probe_spawn_safety(feedback_sink_worker, task_name, group="system") + worker_task( + task_definition_name=task_name, + task_def=_default_task_def(task_name), + register_task_def=True, + overwrite_task_def=True, + domain=domain, + thread_count=_SYSTEM_WORKER_THREADS, + lease_extend_enabled=True, + )(feedback_sink_worker) + def _register_gate_worker( self, agent_name: str, gate_fn, domain: "Optional[str]" = None ) -> None: @@ -2495,6 +2589,9 @@ def run( resolved_prompt = self._check_input_guardrails(agent, resolved_prompt) self._validate_execution_input(resolved_prompt, media=media, context=context) + # OCG long-term memory: inject relevant past memories into the prompt. + agent = self._apply_memory_retrieval(agent, resolved_prompt) + correlation_id = str(uuid.uuid4()) logger.info("Executing agent '%s'", agent.name) @@ -2575,7 +2672,7 @@ def run( error_reason = task_failure_reason or status.reason logger.info("Agent '%s' completed (execution_id=%s)", agent.name, execution_id) - return AgentResult( + result = AgentResult( output=output, execution_id=execution_id, correlation_id=correlation_id, @@ -2588,6 +2685,118 @@ def run( sub_results=self._extract_sub_results(output), ) + # OCG long-term memory: summarize the conversation into a memory and + # hand the human good/bad capability links to the agent's feedback_sink. + self._maybe_save_conversation_memory(agent, result, session_id) + + return result + + # ── OCG long-term memory hooks ───────────────────────────────── + + def _apply_memory_retrieval(self, agent: "Agent", prompt: str) -> "Agent": + """Prepend relevant OCG memories to a COPY of the agent's instructions. + + No-op (returns the original agent) when the agent has no ``semantic_memory`` + or nothing relevant is found. Never mutates the shared agent instance. + """ + sm = getattr(agent, "semantic_memory", None) + if sm is None: + return agent + try: + context = sm.get_context(prompt) + except Exception as exc: # retrieval is best-effort + logger.warning("memory retrieval failed for agent '%s': %s", agent.name, exc) + return agent + if not context: + return agent + + import copy as _copy + + augmented = _copy.copy(agent) + base = agent.instructions if isinstance(agent.instructions, str) else "" + augmented.instructions = (context + "\n\n" + base) if base else context + return augmented + + def _maybe_save_conversation_memory( + self, agent: "Agent", result: "AgentResult", session_id: Optional[str] + ) -> None: + """Summarize the run's conversation into an OCG memory (best-effort). + + Runs an internal summarizer agent (server-side) to distill durable facts, + stores the result via the agent's ``semantic_memory`` store, then mints the + good/bad capability links and passes them to ``agent.feedback_sink``. All + failures are swallowed — memory saving must never fail the primary run. + """ + sm = getattr(agent, "semantic_memory", None) + if sm is None or getattr(result, "status", None) != "COMPLETED": + return + if not getattr(result, "messages", None): + return + try: + from conductor.ai.agents.ocg_memory import ( + FeedbackEvent, + build_memory_summarizer, + ) + + transcript = self._transcript_text(result.messages) + model = getattr(agent, "memory_summary_model", None) or _agent_model_str(agent) + summarizer = build_memory_summarizer(model) + summary_result = self.run(summarizer, transcript) + + summary, facts, tags = _parse_summary_output(summary_result.output) + content = summary + if facts: + content += "\n\nFacts:\n" + "\n".join(f"- {f}" for f in facts) + + key = f"conversation:{session_id or result.execution_id}" + store = sm.store + from conductor.ai.agents.semantic_memory import MemoryEntry + + store.add( + MemoryEntry( + id=key, + content=content, + metadata={"key": key, "tags": list(tags) + ["conversation"]}, + ) + ) + logger.info("saved conversation memory '%s' for agent '%s'", key, agent.name) + + sink = getattr(agent, "feedback_sink", None) + if sink is None: + return + links: Dict[str, Any] = {} + if hasattr(store, "feedback_links"): + try: + links = store.feedback_links(key) + except Exception as exc: + logger.warning("could not mint feedback links for '%s': %s", key, exc) + event = FeedbackEvent( + memory_key=key, + summary=summary, + facts=list(facts), + tags=list(tags), + good_url=links.get("good_url"), + bad_url=links.get("bad_url"), + expires_at=links.get("expires_at"), + agent=getattr(store, "_agent", None), + user=getattr(store, "_user", None), + session_id=session_id, + ) + sink(event) + except Exception as exc: # never fail the primary run + logger.warning("conversation memory save failed for agent '%s': %s", agent.name, exc) + + @staticmethod + def _transcript_text(messages: List[Dict[str, Any]]) -> str: + lines: List[str] = [] + for m in messages: + role = str(m.get("role", "")) or "?" + content = m.get("content", "") + if not isinstance(content, str): + content = json.dumps(content) + lines.append(f"{role}: {content}") + return "\n".join(lines) + # ── Run-by-name (pre-deployed agents) ────────────────────────── def _run_by_name( diff --git a/tests/unit/ai/test_config_serializer.py b/tests/unit/ai/test_config_serializer.py index 404427f0..d2c9bab7 100644 --- a/tests/unit/ai/test_config_serializer.py +++ b/tests/unit/ai/test_config_serializer.py @@ -214,6 +214,69 @@ def test_serialize_memory(self): assert config["memory"]["messages"] == [{"role": "system", "message": "context"}] + def test_serialize_long_term_memory(self): + """OCG-backed semantic_memory serializes to longTermMemory + feedbackSink.""" + from conductor.ai.agents.agent import Agent + from conductor.ai.agents.ocg_memory import OCGMemoryStore + from conductor.ai.agents.semantic_memory import SemanticMemory + + store = OCGMemoryStore( + url="https://ocg.example.com/", + agent="agent:ce-ticket-resolution", + user="user:alice", + scope="agent", + ) + sm = SemanticMemory(store=store, max_results=7) + + def sink(event): # feedback_sink callable + return None + + agent = Agent( + name="ce_agent", + model="openai/gpt-4o", + instructions="Resolve tickets.", + semantic_memory=sm, + memory_summary_model="openai/gpt-4o-mini", + feedback_sink=sink, + ) + config = self.serializer.serialize(agent) + + ltm = config["longTermMemory"] + assert ltm["ocgUrl"] == "https://ocg.example.com" # trailing slash stripped + assert ltm["credential"] == "OCG_PUBLIC_KEY" # server-resolvable name, not token + assert ltm["agent"] == "agent:ce-ticket-resolution" + assert ltm["user"] == "user:alice" + assert ltm["scope"] == "agent" + assert ltm["maxResults"] == 7 + assert ltm["summaryModel"] == "openai/gpt-4o-mini" + + assert config["feedbackSink"] == {"taskName": "ce_agent_feedback_sink"} + + def test_serialize_long_term_memory_absent(self): + """No semantic_memory -> no longTermMemory / feedbackSink keys (no-op).""" + from conductor.ai.agents.agent import Agent + + agent = Agent(name="plain", model="openai/gpt-4o", instructions="Hi.") + config = self.serializer.serialize(agent) + + assert "longTermMemory" not in config + assert "feedbackSink" not in config + + def test_serialize_long_term_memory_summary_model_fallback(self): + """summaryModel falls back to the agent's own model when unset.""" + from conductor.ai.agents.agent import Agent + from conductor.ai.agents.ocg_memory import OCGMemoryStore + from conductor.ai.agents.semantic_memory import SemanticMemory + + store = OCGMemoryStore(url="https://ocg.example.com", agent="agent:x") + sm = SemanticMemory(store=store, max_results=5) + agent = Agent(name="a", model="anthropic/claude", semantic_memory=sm) + config = self.serializer.serialize(agent) + + assert config["longTermMemory"]["summaryModel"] == "anthropic/claude" + # No feedback_sink -> no feedbackSink emitted. + assert "feedbackSink" not in config + def test_serialize_gate_text(self): """TextGate serializes to text_contains config.""" from conductor.ai.agents.agent import Agent diff --git a/tests/unit/ai/test_ocg_memory_store.py b/tests/unit/ai/test_ocg_memory_store.py new file mode 100644 index 00000000..d5710991 --- /dev/null +++ b/tests/unit/ai/test_ocg_memory_store.py @@ -0,0 +1,253 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. See LICENSE file in the project root for details. + +"""Unit tests for OCG-backed memory: the HTTP store adapter, the conversation +summary helpers, and the runtime save/retrieval hooks.""" + +from __future__ import annotations + +import json +from typing import List + +import httpx +import pytest + +from conductor.ai.agents import Agent +from conductor.ai.agents.exceptions import AgentAPIError +from conductor.ai.agents.ocg_memory import FeedbackEvent, OCGMemoryStore +from conductor.ai.agents.result import AgentResult, Status +from conductor.ai.agents.runtime.runtime import ( + AgentRuntime, + _agent_model_str, + _parse_summary_output, +) +from conductor.ai.agents.semantic_memory import MemoryEntry, MemoryStore, SemanticMemory + + +def _store_with(handler) -> OCGMemoryStore: + client = httpx.Client(transport=httpx.MockTransport(handler)) + return OCGMemoryStore(url="https://ocg.test", agent="agent:a", user="user:bob", client=client) + + +class TestOCGMemoryStore: + def test_add_posts_value_field_and_no_confidence(self): + captured = {} + + def handler(req: httpx.Request) -> httpx.Response: + captured["url"] = str(req.url) + captured["body"] = json.loads(req.content) + return httpx.Response(200, json={"key": "k1"}) + + store = _store_with(handler) + key = store.add(MemoryEntry(content="alice prefers email", metadata={"key": "pref"})) + + assert key == "pref" + assert captured["url"].endswith("/api/v1/memories") + body = captured["body"] + assert body["value"] == "alice prefers email" # field is "value", NOT "string_value" + assert "string_value" not in body + assert "confidence" not in body # confidence was removed from the API + assert body["agent"] == "agent:a" and body["user"] == "user:bob" + + def test_search_folds_good_bad_signal_into_content(self): + def handler(req: httpx.Request) -> httpx.Response: + assert str(req.url).endswith("/api/v1/memories/search") + return httpx.Response( + 200, + json={ + "memories": [ + { + "key": "m1", + "value_preview": "use us-east-1", + "good_count": 2, + "bad_count": 1, + "relevance_score": 0.9, + "feedback_notes": [{"verdict": "bad", "reason": "stale region"}], + } + ] + }, + ) + + store = _store_with(handler) + entries = store.search("which region", top_k=5) + assert len(entries) == 1 + assert "[good 2 / bad 1]" in entries[0].content + assert 'bad: "stale region"' in entries[0].content + + def test_feedback_links_hits_mint_route(self): + def handler(req: httpx.Request) -> httpx.Response: + assert str(req.url).split("?")[0].endswith("/api/v1/memories/k1/feedback-links") + return httpx.Response( + 200, + json={ + "good_url": "https://ocg.test/api/v1/feedback/GOOD", + "bad_url": "https://ocg.test/api/v1/feedback/BAD", + "expires_at": "2026-09-01T00:00:00Z", + }, + ) + + store = _store_with(handler) + links = store.feedback_links("k1") + assert links["good_url"].endswith("/feedback/GOOD") + assert links["bad_url"].endswith("/feedback/BAD") + + def test_non_2xx_raises(self): + def handler(req: httpx.Request) -> httpx.Response: + return httpx.Response(500, text="boom") + + store = _store_with(handler) + with pytest.raises(AgentAPIError): + store.add(MemoryEntry(content="x", metadata={"key": "k"})) + + +class _FakeStore(MemoryStore): + """Records add() calls and serves canned feedback links.""" + + def __init__(self): + self.added: List[MemoryEntry] = [] + self._agent = "agent:a" + self._user = "user:bob" + + def add(self, entry: MemoryEntry) -> str: + self.added.append(entry) + return entry.id or "k" + + def search(self, query: str, top_k: int = 5) -> List[MemoryEntry]: + return [] + + def delete(self, memory_id: str) -> bool: + return True + + def clear(self) -> None: + pass + + def list_all(self) -> List[MemoryEntry]: + return [] + + def feedback_links(self, key: str): + return { + "good_url": "https://ocg.test/api/v1/feedback/GOOD", + "bad_url": "https://ocg.test/api/v1/feedback/BAD", + "expires_at": "2026-09-01T00:00:00Z", + } + + +class TestRuntimeMemoryHooks: + def test_save_stores_distilled_summary_and_invokes_sink(self, monkeypatch): + rt = AgentRuntime() + store = _FakeStore() + events: List[FeedbackEvent] = [] + + agent = Agent( + name="support", + model="openai/gpt-4o", + semantic_memory=SemanticMemory(store=store), + feedback_sink=lambda ev: events.append(ev), + ) + + # Stub the nested summarizer run — return distilled facts, NOT the transcript. + def fake_run(summarizer_agent, transcript, **kwargs): + assert summarizer_agent.name == "__memory_summarizer" + return AgentResult( + output={ + "summary": "Alice is on Enterprise.", + "facts": ["plan=enterprise"], + "tags": ["billing"], + }, + status=Status.COMPLETED, + ) + + monkeypatch.setattr(rt, "run", fake_run) + + result = AgentResult( + output={"result": "ok"}, + execution_id="exec-1", + status=Status.COMPLETED, + messages=[{"role": "user", "content": "hi"}, {"role": "assistant", "content": "hello"}], + ) + + rt._maybe_save_conversation_memory(agent, result, session_id="sess-9") + + assert len(store.added) == 1 + saved = store.added[0] + assert saved.id == "conversation:sess-9" + assert "Alice is on Enterprise." in saved.content + assert "plan=enterprise" in saved.content + assert "hello" not in saved.content # the raw transcript is NOT stored + assert "conversation" in saved.metadata["tags"] + + assert len(events) == 1 + assert events[0].good_url.endswith("/feedback/GOOD") + assert events[0].bad_url.endswith("/feedback/BAD") + assert events[0].memory_key == "conversation:sess-9" + + def test_save_skipped_when_no_semantic_memory(self, monkeypatch): + rt = AgentRuntime() + called = {"run": False} + monkeypatch.setattr(rt, "run", lambda *a, **k: called.__setitem__("run", True)) + agent = Agent(name="plain", model="openai/gpt-4o") + rt._maybe_save_conversation_memory( + agent, + AgentResult(status=Status.COMPLETED, messages=[{"role": "user", "content": "hi"}]), + None, + ) + assert called["run"] is False # no nested summarizer run, no recursion + + def test_save_never_raises_on_failure(self, monkeypatch): + rt = AgentRuntime() + + def boom(*a, **k): + raise RuntimeError("summarizer exploded") + + monkeypatch.setattr(rt, "run", boom) + agent = Agent( + name="support", + model="openai/gpt-4o", + semantic_memory=SemanticMemory(store=_FakeStore()), + ) + # Must not raise. + rt._maybe_save_conversation_memory( + agent, + AgentResult(status=Status.COMPLETED, messages=[{"role": "user", "content": "hi"}]), + None, + ) + + def test_apply_retrieval_prepends_context_without_mutating_original(self): + rt = AgentRuntime() + store = _FakeStore() + sm = SemanticMemory(store=store) + sm.get_context = lambda q: "Relevant context from memory:\n 1. plan=enterprise" # type: ignore + agent = Agent( + name="support", model="openai/gpt-4o", instructions="Be helpful.", semantic_memory=sm + ) + + augmented = rt._apply_memory_retrieval(agent, "what plan?") + assert augmented is not agent + assert augmented.instructions.startswith("Relevant context from memory:") + assert "Be helpful." in augmented.instructions + assert agent.instructions == "Be helpful." # original untouched + + def test_apply_retrieval_noop_without_memory(self): + rt = AgentRuntime() + agent = Agent(name="plain", model="openai/gpt-4o", instructions="Hi") + assert rt._apply_memory_retrieval(agent, "q") is agent + + +class TestSummaryHelpers: + def test_agent_model_str_fallback(self): + assert _agent_model_str(Agent(name="a", model="openai/gpt-4o")) == "openai/gpt-4o" + + def test_parse_summary_from_dict(self): + s, f, t = _parse_summary_output({"summary": "x", "facts": ["a"], "tags": ["b"]}) + assert s == "x" and f == ["a"] and t == ["b"] + + def test_parse_summary_from_wrapped_result(self): + s, f, t = _parse_summary_output({"result": {"summary": "y", "facts": [], "tags": []}}) + assert s == "y" + + def test_agent_stores_memory_attrs(self): + sm = SemanticMemory(store=_FakeStore()) + agent = Agent(name="a", model="openai/gpt-4o", semantic_memory=sm) + assert agent.semantic_memory is sm + assert agent.memory_summary_model is None # defaults to None -> reuse agent model + assert agent.feedback_sink is None