diff --git a/docs/agents/reference/agent-schema.json b/docs/agents/reference/agent-schema.json index 12656b47..414e1be9 100644 --- a/docs/agents/reference/agent-schema.json +++ b/docs/agents/reference/agent-schema.json @@ -19,6 +19,17 @@ "properties": { "taskName": { "type": "string", "minLength": 1 } }, "additionalProperties": false }, + "longTermMemory": { + "type": "object", + "required": ["ocgUrl", "credential", "agent"], + "properties": { + "ocgUrl": { "type": "string", "minLength": 1 }, + "credential": { "type": "string", "minLength": 1 }, + "agent": { "type": "string", "minLength": 1 }, + "user": { "type": "string" } + }, + "additionalProperties": false + }, "tool": { "type": "object", "required": ["name"], @@ -52,6 +63,7 @@ "outputType": { "type": "object" }, "guardrails": { "type": "array", "items": { "type": "object" } }, "memory": { "type": "object" }, + "longTermMemory": { "$ref": "#/$defs/longTermMemory" }, "maxTokens": { "type": "integer", "minimum": 0 }, "contextWindowBudget": { "type": "integer", "minimum": 0 }, "temperature": { "type": "number" }, diff --git a/e2e/test_suite22_ocg.py b/e2e/test_suite22_ocg.py deleted file mode 100644 index 1d0a7c04..00000000 --- a/e2e/test_suite22_ocg.py +++ /dev/null @@ -1,200 +0,0 @@ -"""Suite 22: OCG multi-instance — per-tool instance binding isolation. - -The multi-tenancy guarantee of the SDK-defined OCG design: two retrieval -agents bound to two different OCG instances (`ocg_agent(url=...)`) each hit -their own instance and ONLY that instance. Validation is purely structural — -recorded HTTP traffic on the stubs — never LLM-judged output quality. - - 1. US agent (agent_tool → ocg_agent bound to stub A) → traffic on A, none on B - 2. Canada agent (bound to stub B) → traffic on B, none on A - 3. Negative: agent with no OCG tools → no traffic on either stub - -Manages two stub OCG instances on dedicated ports. -No mocks of Conductor Agents itself. Real server, real LLM, stub OCG backends. -""" - -import json -import os -import threading -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer - -import pytest - -from conductor.ai.agents import Agent, agent_tool -from conductor.ai.agents.ocg import ocg_agent - -pytestmark = [ - pytest.mark.e2e, - pytest.mark.xdist_group("ocg"), -] - -# ── Configuration ──────────────────────────────────────────────────────── - -US_PORT = 3061 -CA_PORT = 3062 -TIMEOUT = 120 - -MODEL = os.environ.get("CONDUCTOR_AGENT_LLM_MODEL", "openai/gpt-4o-mini") - -# The Conductor Agents server resolves the per-tool OCG URL server-side, so the -# stubs must be reachable from the server process — localhost works for the -# local e2e topology (server and tests on the same host). -US_URL = f"http://localhost:{US_PORT}" -CA_URL = f"http://localhost:{CA_PORT}" - - -# ── Stub OCG instance ──────────────────────────────────────────────────── - - -class _StubOcg: - """Minimal OCG lookalike: answers /api/v1/agent/query with canned - citations and records every request it receives.""" - - def __init__(self, port: int, region: str): - self.port = port - self.region = region - self.requests: list = [] # (method, path, body) tuples - stub = self - - class Handler(BaseHTTPRequestHandler): - def _record_and_reply(self, body: str): - stub.requests.append((self.command, self.path, body)) - payload = { - "citations": [ - { - "source_item_id": f"{stub.region}-item-1", - "title": f"{stub.region} maintenance window", - "container_id": f"#{stub.region}-ops", - "snippet": f"The {stub.region} maintenance window is Saturday 02:00 UTC.", - } - ] - } - data = json.dumps(payload).encode() - self.send_response(200) - self.send_header("Content-Type", "application/json") - self.send_header("Content-Length", str(len(data))) - self.end_headers() - self.wfile.write(data) - - def do_POST(self): - length = int(self.headers.get("Content-Length", 0)) - self._record_and_reply(self.rfile.read(length).decode()) - - def do_GET(self): - self._record_and_reply("") - - def do_DELETE(self): - self._record_and_reply("") - - def log_message(self, *args): # silence per-request stderr noise - pass - - self._server = ThreadingHTTPServer(("0.0.0.0", port), Handler) - self._thread = threading.Thread(target=self._server.serve_forever, daemon=True) - - def start(self): - self._thread.start() - return self - - def stop(self): - self._server.shutdown() - self._server.server_close() - - @property - def query_requests(self): - return [r for r in self.requests if r[1].startswith("/api/v1/agent/query")] - - -@pytest.fixture(scope="module") -def stubs(): - us = _StubOcg(US_PORT, "us").start() - ca = _StubOcg(CA_PORT, "canada").start() - try: - yield us, ca - finally: - us.stop() - ca.stop() - - -def _retrieval_main(name: str, retriever) -> Agent: - return Agent( - name=name, - model=MODEL, - instructions=( - "You answer operational questions. You MUST call your retrieval " - "tool to look up the answer before responding — never answer " - "from memory and never ask the user clarifying questions. Pass " - "the user's question to the retrieval tool verbatim." - ), - tools=[agent_tool(retriever)], - max_turns=6, - ) - - -PROMPT = ( - "Search for recent messages about the maintenance window for cluster " - "prod-east and report exactly what the messages say. Do not ask " - "clarifying questions — search first." -) - - -# ── Tests ──────────────────────────────────────────────────────────────── - - -@pytest.mark.timeout(TIMEOUT * 2) -def test_us_agent_hits_only_us_instance(runtime, stubs): - us, ca = stubs - us_before, ca_before = len(us.query_requests), len(ca.query_requests) - - retriever = ocg_agent(name="ocg_us_e2e", model=MODEL, url=US_URL) - main = _retrieval_main("ocg_e2e_us_main", retriever) - - result = runtime.run(main, PROMPT, timeout=TIMEOUT) - assert result is not None - - # The multi-tenancy guarantee, asserted on recorded traffic: - assert len(us.query_requests) > us_before, ( - f"US-bound retriever never queried the US OCG stub — stub saw: {us.requests}" - ) - assert len(ca.query_requests) == ca_before, ( - f"US-bound retriever leaked traffic to the Canada stub: {ca.requests}" - ) - - -@pytest.mark.timeout(TIMEOUT * 2) -def test_canada_agent_hits_only_canada_instance(runtime, stubs): - us, ca = stubs - us_before, ca_before = len(us.query_requests), len(ca.query_requests) - - retriever = ocg_agent(name="ocg_ca_e2e", model=MODEL, url=CA_URL) - main = _retrieval_main("ocg_e2e_ca_main", retriever) - - result = runtime.run(main, PROMPT, timeout=TIMEOUT) - assert result is not None - - assert len(ca.query_requests) > ca_before, ( - f"Canada-bound retriever never queried the Canada OCG stub — stub saw: {ca.requests}" - ) - assert len(us.query_requests) == us_before, ( - f"Canada-bound retriever leaked traffic to the US stub: {us.requests}" - ) - - -@pytest.mark.timeout(TIMEOUT * 2) -def test_agent_without_ocg_tools_generates_no_ocg_traffic(runtime, stubs): - us, ca = stubs - us_before, ca_before = len(us.requests), len(ca.requests) - - plain = Agent( - name="ocg_e2e_plain", - model=MODEL, - instructions="Answer briefly from your own knowledge.", - max_turns=2, - ) - - result = runtime.run(plain, "Say hello in one word.", timeout=TIMEOUT) - assert result is not None - - # Inverse of the deleted auto-expose behavior: no OCG opt-in, no OCG calls. - assert len(us.requests) == us_before - assert len(ca.requests) == ca_before diff --git a/examples/agents/116_ocg_subagent.py b/examples/agents/116_ocg_subagent.py deleted file mode 100644 index a90c8712..00000000 --- a/examples/agents/116_ocg_subagent.py +++ /dev/null @@ -1,86 +0,0 @@ -"""116 — OCG retrieval via the prebuilt sub-agent. - -The main agent delegates retrieval to an OCG (Open Context Graph) -sub-agent: ``ocg_agent()`` returns an ordinary ``Agent`` carrying the -canned retrieval prompt and all seven ``ocg_*`` tools; wrapping it with -``agent_tool()`` exposes it to the main agent's LLM as a single tool. - -When the main agent calls it, the sub-agent runs its *own* LLM loop — -it can issue several OCG queries and walk entity neighborhoods — and -returns one synthesized, cited answer. The main agent's -context only ever sees that final answer, not the raw graph payloads. - -Choose this shape when retrieval takes judgment (multi-step lookups, -aggregation in two steps, query reformulation). For a single direct -lookup from the main agent's own loop, see -``117_ocg_direct_tools.py``. - -OCG is opt-in per agent — nothing is auto-injected, and every OCG tool -binds the instance it talks to (no server-side default): set -``OCG_INSTANCE_URL`` (and optionally ``OCG_CREDENTIAL``, a -credential-store *name*). - -Run (from ``sdk/python``):: - - # one-time: store the OCG bearer token in the server's secrets store, - # e.g. in orkes: PUT /api/secrets/OCG_PUBLIC_KEY '""' - - OCG_INSTANCE_URL=https://test.contextgraph.io \ - OCG_CREDENTIAL=OCG_PUBLIC_KEY \ - uv run python examples/116_ocg_subagent.py - - # against an embedded server (e.g. orkes on 8080), add: - # CONDUCTOR_SERVER_URL=http://localhost:8080/api -""" - -import os - -from conductor.ai.agents import Agent, AgentRuntime, agent_tool -from conductor.ai.agents.ocg import ocg_agent - -MODEL = os.environ.get("CONDUCTOR_AGENT_LLM_MODEL", "anthropic/claude-sonnet-4-6") - -# Per-tool instance binding — required: every OCG tool binds the instance -# it talks to; there is no server-side default. -OCG_INSTANCE_URL = os.environ.get("OCG_INSTANCE_URL") or "" -OCG_CREDENTIAL = os.environ.get("OCG_CREDENTIAL") # credential-store name, never the key -if not OCG_INSTANCE_URL: - raise SystemExit("Set OCG_INSTANCE_URL to your OCG instance, e.g. https://test.contextgraph.io") - -PROMPT = ( - "Catch me up on 'Improvements to Python SDK -- performance, Feature " - "parity, logging, metrics etc'. What's the current state, what's " - "underneath it, and what's been changing in the codebase?" -) - - -def main() -> None: - retriever = ocg_agent( - name="ocg_retriever", - model=MODEL, - url=OCG_INSTANCE_URL, - credential=OCG_CREDENTIAL, - ) - - main_agent = Agent( - name="jira_ocg_subagent", - model=MODEL, - instructions=( - "You answer questions about the team's work. Call your " - "retrieval tool exactly once, passing the user's full " - "question — messages and Jira tickets all live " - "behind it. Its answer is complete: when it returns, write " - "your final response as a concise brief of what it found, " - "keeping its citations." - ), - tools=[agent_tool(retriever)], - max_turns=4, - ) - - with AgentRuntime() as runtime: - result = runtime.run(main_agent, PROMPT) - result.print_result() - - -if __name__ == "__main__": - main() diff --git a/examples/agents/117_ocg_direct_tools.py b/examples/agents/117_ocg_direct_tools.py deleted file mode 100644 index bf2527f8..00000000 --- a/examples/agents/117_ocg_direct_tools.py +++ /dev/null @@ -1,85 +0,0 @@ -"""117 — OCG retrieval via a direct tool call (no sub-agent). - -The main agent holds the OCG query tool *itself*: ``ocg_tools()`` -returns raw ``ToolDef``s that dispatch straight to the server's -``OCG_*`` system tasks, so the main agent's own LLM issues the query -and reads the citations — no sub-agent hop, no second LLM loop. - -Compared to ``116_ocg_subagent.py``: - -- one LLM round-trip cheaper per lookup — there is no retrieval agent - spending its own turns; -- the raw (projected, capped) OCG response lands directly in the main - agent's context, so IT does the reading — fine for a single focused - query, wasteful when retrieval takes several exploratory calls; -- you own the retrieval prompting: the canned OCG system prompt is the - sub-agent's, so any query-writing guidance the model needs (specific - keywords, time bounds, two-step aggregation) belongs in your own - ``instructions`` here. - -This example exposes only ``ocg_query`` (the subset switches turn off -entity/memory tools) — the narrowest possible OCG surface. - -Instance binding works exactly as in 116: ``OCG_INSTANCE_URL`` (required) / -``OCG_CREDENTIAL`` env vars. - -Run (from ``sdk/python``):: - - OCG_INSTANCE_URL=https://test.contextgraph.io \ - OCG_CREDENTIAL=OCG_PUBLIC_KEY \ - uv run python examples/117_ocg_direct_tools.py - - # against an embedded server (e.g. orkes on 8080), add: - # CONDUCTOR_SERVER_URL=http://localhost:8080/api -""" - -import os - -from conductor.ai.agents import Agent, AgentRuntime -from conductor.ai.agents.ocg import ocg_tools - -MODEL = os.environ.get("CONDUCTOR_AGENT_LLM_MODEL", "anthropic/claude-sonnet-4-6") - -OCG_INSTANCE_URL = os.environ.get("OCG_INSTANCE_URL") or "" -OCG_CREDENTIAL = os.environ.get("OCG_CREDENTIAL") # credential-store name, never the key -if not OCG_INSTANCE_URL: - raise SystemExit("Set OCG_INSTANCE_URL to your OCG instance, e.g. https://test.contextgraph.io") - -PROMPT = ( - "Catch me up on 'Improvements to Python SDK -- performance, Feature " - "parity, logging, metrics etc'. What's the current state, what's " - "underneath it, and what's been changing in the codebase?" -) - - -def main() -> None: - main_agent = Agent( - name="jira_ocg_direct", - model=MODEL, - instructions=( - "You answer questions about the team's work using ocg_query, " - "a keyword/embedding retrieval tool (NOT an LLM) over a " - "knowledge graph of messages and Jira tickets. Query " - "with specific keywords (ticket titles, component names) — " - "under ~15 content words, never phrased as a question. At " - "most one query per topic, 4 total; never repeat or rephrase " - "a query. When the queries are done, write your final " - "response: a concise brief synthesized from the citations." - ), - max_turns=6, - tools=ocg_tools( - url=OCG_INSTANCE_URL, - credential=OCG_CREDENTIAL, - query=True, - entities=False, - memory=False, - ), - ) - - with AgentRuntime() as runtime: - result = runtime.run(main_agent, PROMPT) - result.print_result() - - -if __name__ == "__main__": - main() diff --git a/src/conductor/ai/agents/__init__.py b/src/conductor/ai/agents/__init__.py index fce5b308..8d0f7f96 100644 --- a/src/conductor/ai/agents/__init__.py +++ b/src/conductor/ai/agents/__init__.py @@ -174,8 +174,8 @@ def resolve_credentials(task: object, names: list) -> dict: # Agent discovery -# OCG (Open Context Graph) retrieval sub-agent -from conductor.ai.agents.ocg import OCG_SYSTEM_PROMPT, ocg_agent, ocg_tools +# OCG configuration +from conductor.ai.agents.ocg_config import OcgConfig # OpenAI Agents SDK compatibility from conductor.ai.agents.openai_compat import Runner, RunResult @@ -257,10 +257,8 @@ def resolve_credentials(task: object, names: list) -> dict: "agent_tool", "api_tool", "http_tool", - # OCG retrieval sub-agent - "OCG_SYSTEM_PROMPT", - "ocg_agent", - "ocg_tools", + # OCG configuration + "OcgConfig", "human_tool", "mcp_tool", "wait_for_message_tool", diff --git a/src/conductor/ai/agents/agent.py b/src/conductor/ai/agents/agent.py index b199aa11..a05c2485 100644 --- a/src/conductor/ai/agents/agent.py +++ b/src/conductor/ai/agents/agent.py @@ -15,6 +15,7 @@ from typing import Any, Callable, Dict, List, Optional, Union from conductor.ai.agents.claude_code import ClaudeCode +from conductor.ai.agents.ocg_config import OcgConfig _VALID_NAME_RE = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_-]*$") @@ -501,6 +502,7 @@ class Agent: output_type: A Pydantic model or dataclass for structured output. guardrails: List of :class:`Guardrail` instances for input/output validation. memory: Optional :class:`ConversationMemory` for session management. + ocg: Optional :class:`OcgConfig` for Conductor-managed OCG integration. dependencies: Optional dict of dependencies to inject into tool context. max_turns: Maximum agent loop iterations (default 25). max_tokens: Maximum tokens for LLM generation. @@ -559,6 +561,7 @@ def __init__( output_type: Optional[type] = None, guardrails: Optional[List[Any]] = None, memory: Optional[Any] = None, + ocg: Optional[OcgConfig] = None, dependencies: Optional[Dict[str, Any]] = None, max_turns: int = 25, max_tokens: Optional[int] = None, @@ -721,6 +724,7 @@ def __init__( self.output_type = output_type self.guardrails: List[Any] = list(guardrails) if guardrails else [] self.memory = memory + self.ocg = ocg 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 fcc8ed67..049b0daf 100644 --- a/src/conductor/ai/agents/config_serializer.py +++ b/src/conductor/ai/agents/config_serializer.py @@ -131,6 +131,18 @@ def _serialize_agent(self, agent: "Agent") -> dict: if hasattr(agent, "memory") and agent.memory: config["memory"] = self._serialize_memory(agent.memory) + # OCG behavior is server-managed. The SDK only supplies configuration + # metadata and never performs local retrieval or post-run processing. + if agent.ocg is not None: + long_term_memory: Dict[str, Any] = { + "ocgUrl": agent.ocg.url, + "credential": agent.ocg.credential, + "agent": f"conductor-agent:{agent.name}", + } + if agent.ocg.user is not None: + long_term_memory["user"] = agent.ocg.user + config["longTermMemory"] = long_term_memory + # Max tokens if agent.max_tokens is not None: config["maxTokens"] = agent.max_tokens diff --git a/src/conductor/ai/agents/ocg.py b/src/conductor/ai/agents/ocg.py deleted file mode 100644 index c80ce268..00000000 --- a/src/conductor/ai/agents/ocg.py +++ /dev/null @@ -1,448 +0,0 @@ -"""OCG (Open Context Graph) retrieval sub-agent. - -OCG is a retrieval engine over a knowledge graph of entities (messages, -channels, people) linked by claims and relationships. This module is -the canonical — and only — definition of the OCG integration: system -prompt, tool schemas, endpoint routing, and instance binding all live -here. The tools compile to plain Conductor HTTP tasks (with path -templating); there is no OCG-specific server code at all. - -Typical usage — delegate retrieval from a main agent:: - - from conductor.ai.agents import Agent, agent_tool - from conductor.ai.agents.ocg import ocg_agent - - retriever = ocg_agent(model="anthropic/claude-sonnet-4-6", - url="https://ocg.example.com", - credential="OCG_KEY") - main = Agent(name="support", model="openai/gpt-4o", - tools=[agent_tool(retriever)], instructions="...") - -Multi-instance (e.g. data residency) — bind each retriever to its own OCG:: - - us = ocg_agent(name="ocg_us", model="anthropic/claude-sonnet-4-6", - url="https://us.ocg.example.com", credential="OCG_US_KEY") - ca = ocg_agent(name="ocg_canada", model="anthropic/claude-sonnet-4-6", - url="https://ca.ocg.example.com", credential="OCG_CA_KEY") - -``url`` is required — every OCG tool set binds the instance it talks to; -there is no server-side default. ``credential`` names an entry in the -server's credential store — the secret itself never appears in Python code -or serialized configs. - -.. warning:: - Agents bound to **different** OCG instances must have **distinct** - ``name``s: inline ``agent_tool`` child workflows are registered by agent - name, so two differently-configured agents sharing a name overwrite each - other's workflow definition. -""" - -from __future__ import annotations - -from typing import TYPE_CHECKING, Any, Dict, List, Optional - -from conductor.ai.agents.tool import ToolDef - -if TYPE_CHECKING: - from conductor.ai.agents.agent import Agent - -# ── System prompt ─────────────────────────────────────────────────────── -# -# ``${workflow.input.__today__}`` is substituted by Conductor when the LLM -# task is scheduled — the agent_tool dispatch script injects ``__today__`` -# with the current UTC date on every call, so relative-date queries anchor -# on the real current date instead of one baked in at definition time. - -OCG_SYSTEM_PROMPT = """\ -Today's date is ${workflow.input.__today__} (UTC). DEFAULT: do NOT set -start_time or end_time — OMIT BOTH ENTIRELY. Searching the full history is -the norm, and an unrequested time range silently drops older context that -is usually exactly what you need. ONLY add a time range when the user -EXPLICITLY asks about a recent or time-bounded window ("recent", "last -week", "since Friday", "in May"); then anchor on today's date and set -start_time (and set end_time only for a window that closed in the past). -Timestamps must be full RFC3339 (2026-06-04T00:00:00Z); a bare date is -rejected. Never invent a range the user did not ask for. - -You are querying an OCG (Open Context Graph). It is a RETRIEVAL -engine over a knowledge graph of entities (messages, channels, people) -linked by claims and relationships — embedding/keyword search, NOT an LLM. -It is NOT an aggregation engine and NOT a conversation partner: rephrasing -the same intent returns the same results. - -RETRIEVAL BUDGET: make at most 3 queries total, each with a genuinely -DIFFERENT keyword set. Never repeat or lightly rephrase a query. When the -budget is spent — or results start repeating — STOP querying and answer -from what you have. Timestamps must be full RFC3339 -(2026-06-04T00:00:00Z); a bare date is rejected. - -It can answer: - - "Find messages in channel X about Y" - - "Show TIMED_OUT errors for cluster " - - "What entities mention 'health check failure'?" - - "Recent messages in #cloud_saas_health_check_alerts" - -It CANNOT directly answer (you must do it yourself in two steps): - - "How many of X are there?" / "Which X is most frequent?" - - "Group these by Y" / "Top N by count" - - Statistical or comparative questions - -RESPONSE SIZE: ALWAYS request max_results=100 — it is both the maximum and -the floor for getting decent context; NEVER use a small value like 10 or -25, which starves your answer. ALWAYS set traversal_level = 1 (never 0, -never higher). To focus results, sharpen the KEYWORDS — never by lowering -max_results or by adding an unrequested time range. - -DIG DEEPER: the first ocg_query is only your entry point. After it returns, -pick the 1-3 MOST RELEVANT entities from the citations (the ones most on -point for the question) and call ocg_neighborhood on each — using the -entity ids from the citation rows — to pull in their linked entities -(related tickets, incidents, sub-workflows, prior fixes). The actual fix -very often lives one hop away in a linked entity, not in the first page of -citations. Do not answer from the initial citations alone when a clearly -relevant entity is worth expanding. - -For aggregation questions, use a TWO-STEP pattern: - 1. RETRIEVE: ask OCG for the relevant entities. - - Use specific terms (cluster names, error codes, channel names). - - Use start_time (and end_time only for windows closed in the - past) to bound the range. - - Avoid hedging words ("frequently", "across", "occurrences") — - OCG ranks by keyword presence, and these are noise tokens. - 2. AGGREGATE: count, group, rank yourself from the citation list. - -Query length: keep it under ~15 content words. Long prompts dilute the -BM25 keyword set; OCG's parser is extracting things like "happen", -"identify", "top one" which are not real signal. - -Bad: "Across all clusters, what alert/notification/error type appears - most frequently? Group similar alerts and tell me which one has - the highest count and how many clusters it affected." - -Good (step 1): { - "query": "TIMED_OUT health check failure cluster", - "max_results": 100 -} -(no start_time/end_time — the query runs across the full history.) -Then parse the returned citations, extract cluster names from titles, -build the frequency table in your reasoning.""" - - -# ── JSON-schema helpers ───────────────────────────────────────────────── - - -def _prop(type_: str, description: str, default: Any = None) -> Dict[str, Any]: - schema: Dict[str, Any] = {"type": type_, "description": description} - if default is not None: - schema["default"] = default - return schema - - -def _array(item_type: str, description: str) -> Dict[str, Any]: - return {"type": "array", "description": description, "items": {"type": item_type}} - - -def _object(properties: Dict[str, Any], required: List[str]) -> Dict[str, Any]: - schema: Dict[str, Any] = {"type": "object", "properties": properties} - if required: - schema["required"] = required - return schema - - -# ── Tool definitions ──────────────────────────────────────────────────── - - -def _query_tool() -> Dict[str, Any]: - return { - "name": "ocg_query", - "method": "POST", - "path": "/api/v1/agent/query", - "description": ( - "Query the Open Context Graph for structured retrieval. " - "Returns citations (source_item_id, title, container_id, snippet) " - "and traversal_results when traversal_level > 0." - ), - "schema": _object( - { - "query": _prop("string", "Natural-language retrieval query."), - "max_results": { - "type": "integer", - "description": "Max citations to return. ALWAYS use 100 — it is both " - "the hard maximum and the floor for getting decent context; never " - "request fewer.", - "default": 100, - "minimum": 100, - "maximum": 100, - }, - "traversal_level": { - "type": "integer", - "description": "ALWAYS 1 — pulls each citation's immediate neighborhood " - "in alongside it. Never 0 (too shallow) and never higher (dig deeper by " - "calling ocg_neighborhood on the most relevant entities, not by raising " - "this).", - "default": 1, - "minimum": 1, - "maximum": 1, - }, - "start_time": _prop( - "string", - "LEAVE UNSET by default — omit it so the search covers the FULL " - "history. Set this ONLY when the user EXPLICITLY asked about a recent " - "or time-bounded window (e.g. 'last week', 'since Friday'). RFC3339 " - "lower bound (inclusive), e.g. 2026-06-04T00:00:00Z; a bare date like " - "2026-06-04 is REJECTED.", - ), - "end_time": _prop( - "string", - "LEAVE UNSET by default. Set this ONLY for a window the user said has " - "CLOSED in the past; for anything running through now, omit it. RFC3339 " - "upper bound (exclusive), e.g. 2026-06-11T00:00:00Z; a bare date is " - "REJECTED.", - ), - }, - ["query"], - ), - } - - -def _get_entity_tool() -> Dict[str, Any]: - return { - "name": "ocg_get_entity", - "method": "GET", - "path": "/api/v1/entities/{entity_id}", - "description": "Fetch one entity by its canonical id.", - "schema": _object( - {"entity_id": _prop("string", "Canonical entity id from an ocg_query result row.")}, - ["entity_id"], - ), - } - - -def _neighborhood_tool() -> Dict[str, Any]: - return { - "name": "ocg_neighborhood", - "method": "GET", - "path": "/api/v1/graph/neighborhood/{entity_id}", - "query_params": ["depth", "limit"], - "description": ( - "Get an entity plus its graph neighbors out to `depth` hops. " - "Use limit <= 10, depth=1 on the first call — well-connected " - "entities can have many edges and large responses will be truncated." - ), - "schema": _object( - { - "entity_id": _prop("string", "Entity at the center of the neighborhood."), - "depth": _prop("integer", "Hop depth (use depth=1 on first call).", 1), - "limit": _prop( - "integer", "Cap on neighbors returned (use <= 10 on first call).", 50 - ), - }, - ["entity_id"], - ), - } - - -def _memory_set_tool() -> Dict[str, Any]: - return { - "name": "ocg_memory_set", - "method": "POST", - "path": "/api/v1/memories", - "description": ( - "Create or overwrite a memory in OCG. Cap inferred confidence at 0.7; " - "never write PII or secrets." - ), - "schema": _object( - { - "key": _prop("string", "Memory key."), - "agent": _prop("string", 'Agent owner (e.g. "agent:").'), - "user": _prop("string", 'User owner (e.g. "user:").'), - "string_value": _prop("string", "Stored value."), - "description": _prop("string", "Human-readable description."), - "scope": _prop( - "string", - "Memory scope. One of MEMORY_SCOPE_SESSION, MEMORY_SCOPE_AGENT, " - "MEMORY_SCOPE_USER, MEMORY_SCOPE_SHARED, MEMORY_SCOPE_GLOBAL.", - "MEMORY_SCOPE_USER", - ), - "confidence": _prop("number", "Inferred confidence in [0,1]. Cap at 0.7.", 0.7), - "source_ref": _prop("string", "Free-form source reference (e.g. message id)."), - "evidence_ids": _array("string", "Supporting evidence entity ids."), - "tags": _array("string", "Tags."), - "expires_at": _prop("string", "ISO-8601 expiry. Optional — default 180 days."), - "idempotency_key": _prop("string", "Idempotency key. Optional."), - }, - ["key", "agent", "user", "string_value", "description"], - ), - } - - -def _memory_reinforce_tool() -> Dict[str, Any]: - return { - "name": "ocg_memory_reinforce", - "method": "POST", - "path": "/api/v1/memories/{key}/reinforce", - "description": ( - "Reinforce an existing memory on independent re-observation. confidence_boost must be <= 0.05." - ), - "schema": _object( - { - "key": _prop("string", "Memory key."), - "agent": _prop("string", "Agent owner."), - "user": _prop("string", "User owner."), - "confidence_boost": _prop( - "number", "Boost to add (must be <= 0.05 to prevent compounding drift).", 0.05 - ), - "source_ref": _prop("string", "Free-form source reference."), - }, - ["key", "agent", "user"], - ), - } - - -def _memory_delete_tool() -> Dict[str, Any]: - return { - "name": "ocg_memory_delete", - "method": "DELETE", - "path": "/api/v1/memories/{key}", - "query_params": ["agent", "user"], - "description": ( - "Delete a memory by key. Prefer ocg_memory_set with a corrected value " - "over deletion (preserves history)." - ), - "schema": _object( - { - "key": _prop("string", "Memory key."), - "agent": _prop("string", "Agent owner."), - "user": _prop("string", "User owner."), - }, - ["key", "agent", "user"], - ), - } - - -# ── Public factories ──────────────────────────────────────────────────── - - -def ocg_tools( - *, - url: str, - credential: Optional[str] = None, - query: bool = True, - entities: bool = True, - memory: bool = True, -) -> List[ToolDef]: - """Build the raw OCG :class:`ToolDef` list for a custom retrieval agent. - - Each tool is a plain Conductor HTTP task: the compile bakes the - instance URL, endpoint path template, and auth header into the tool's - dispatch config, and the LLM's arguments fill the path/query/body at - call time. No OCG-specific code runs server-side. - - Args: - url: Base URL of the OCG instance this tool set targets. Required — - there is no server-side default instance. - credential: Name of a credential-store entry holding the OCG bearer - token. The server resolves it at execution time — the secret - never appears in the serialized config. - query: Include ``ocg_query``. - entities: Include ``ocg_get_entity`` + ``ocg_neighborhood``. - memory: Include ``ocg_memory_set`` / ``ocg_memory_reinforce`` / - ``ocg_memory_delete``. - - Raises: - ValueError: If ``url`` is blank. - """ - if not url or not url.strip(): - raise ValueError( - "ocg_tools() requires a non-blank url: every OCG tool set binds its own instance." - ) - - base_url = url.strip().rstrip("/") - headers: Dict[str, str] = {} - credentials: List[str] = [] - if credential: - # Standard http-tool placeholder — resolved server-side from the - # credential store at execution; the token never appears here. - headers["Authorization"] = "Bearer ${" + credential + "}" - credentials = [credential] - - selected: List[Dict[str, Any]] = [] - if query: - selected.append(_query_tool()) - if entities: - selected.append(_get_entity_tool()) - selected.append(_neighborhood_tool()) - if memory: - selected.append(_memory_set_tool()) - selected.append(_memory_reinforce_tool()) - selected.append(_memory_delete_tool()) - - tools: List[ToolDef] = [] - for spec in selected: - config: Dict[str, Any] = { - "url": base_url, - "method": spec["method"], - "pathTemplate": spec["path"], - } - if spec.get("query_params"): - config["queryParams"] = list(spec["query_params"]) - if headers: - config["headers"] = dict(headers) - tools.append( - ToolDef( - name=spec["name"], - description=spec["description"], - input_schema=spec["schema"], - tool_type="http", - config=config, - credentials=list(credentials), - ) - ) - return tools - - -def ocg_agent( - *, - model: str, - url: str, - name: str = "ocg_agent", - credential: Optional[str] = None, - instructions: Optional[str] = None, - max_turns: int = 10, - query: bool = True, - entities: bool = True, - memory: bool = True, -) -> "Agent": - """Build the prebuilt OCG retrieval :class:`Agent`. - - Returns an ordinary :class:`Agent` — wrap it with :func:`agent_tool` to - let a main agent delegate retrieval, or use it as a pipeline stage to - retrieve before the main agent runs. - - Args: - model: LLM for the retrieval agent's own turns (required — the right - model depends on cost/latency targets and the OCG corpus). - url: OCG instance base URL (required — no server-side default). - name: Agent name. **Must be distinct per OCG instance** — child - workflows are registered by agent name (see module warning). - credential: Credential-store entry for the instance's bearer token. - instructions: Override the canned :data:`OCG_SYSTEM_PROMPT`. - max_turns: Retrieval loop budget. - query / entities / memory: Tool subset switches, forwarded to - :func:`ocg_tools`. - """ - from conductor.ai.agents.agent import Agent - - return Agent( - name=name, - model=model, - instructions=instructions if instructions is not None else OCG_SYSTEM_PROMPT, - tools=ocg_tools( - url=url, - credential=credential, - query=query, - entities=entities, - memory=memory, - ), - max_turns=max_turns, - ) diff --git a/src/conductor/ai/agents/ocg_config.py b/src/conductor/ai/agents/ocg_config.py new file mode 100644 index 00000000..6eda4f39 --- /dev/null +++ b/src/conductor/ai/agents/ocg_config.py @@ -0,0 +1,38 @@ +# Copyright (c) 2025 Agentspan +# Licensed under the MIT License. See LICENSE file in the project root for details. + +"""Declarative Open Context Graph configuration.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + + +@dataclass(frozen=True) +class OcgConfig: + """Configure Conductor-managed OCG integration for an agent. + + ``credential`` is the name of a Conductor secret. Raw API keys are not + accepted or stored by this configuration. + """ + + url: str + credential: str = "OCG_PUBLIC_KEY" + user: Optional[str] = None + + def __post_init__(self) -> None: + if not isinstance(self.url, str): + raise ValueError("OcgConfig url must be a non-empty string") + normalized_url = self.url.strip().rstrip("/") + if not normalized_url: + raise ValueError("OcgConfig url must be non-empty") + + if not isinstance(self.credential, str): + raise ValueError("OcgConfig credential must be a non-empty secret name") + normalized_credential = self.credential.strip() + if not normalized_credential: + raise ValueError("OcgConfig credential must be a non-empty secret name") + + object.__setattr__(self, "url", normalized_url) + object.__setattr__(self, "credential", normalized_credential) diff --git a/tests/unit/ai/test_agent_schema_contract.py b/tests/unit/ai/test_agent_schema_contract.py index eecd1db3..c77e62ae 100644 --- a/tests/unit/ai/test_agent_schema_contract.py +++ b/tests/unit/ai/test_agent_schema_contract.py @@ -8,6 +8,7 @@ import jsonschema import pytest +from conductor.ai.agents import OcgConfig from conductor.ai.agents.agent import Agent from conductor.ai.agents.config_serializer import AgentConfigSerializer from conductor.ai.agents.guardrail import RegexGuardrail @@ -37,6 +38,7 @@ def test_agent_schema_is_valid_and_accepts_representative_serializer_output(): strategy="handoff", guardrails=[RegexGuardrail(name="safe", patterns=[".*"])], termination=TextMentionTermination(text="DONE"), + ocg=OcgConfig(url="https://ocg.example.com"), metadata={"owner": "docs"}, max_tokens=100, temperature=0.1, diff --git a/tests/unit/ai/test_ocg.py b/tests/unit/ai/test_ocg.py deleted file mode 100644 index 106eea9f..00000000 --- a/tests/unit/ai/test_ocg.py +++ /dev/null @@ -1,201 +0,0 @@ -"""Unit tests for the OCG (Open Context Graph) retrieval sub-agent factories.""" - -import pytest - -from conductor.ai.agents.ocg import OCG_SYSTEM_PROMPT, ocg_agent, ocg_tools -from conductor.ai.agents.tool import ToolDef - -ALL_TOOL_NAMES = { - "ocg_query", - "ocg_get_entity", - "ocg_neighborhood", - "ocg_memory_set", - "ocg_memory_reinforce", - "ocg_memory_delete", -} - - -URL = "https://us.ocg.example.com" - - -class TestOcgTools: - def test_default_returns_all_six(self): - tools = ocg_tools(url=URL) - assert len(tools) == 6 - assert {t.name for t in tools} == ALL_TOOL_NAMES - # OCG tools ARE http tools — they execute as plain Conductor HTTP - # tasks; there is no OCG-specific server code. - assert all(t.tool_type == "http" for t in tools) - assert all(isinstance(t, ToolDef) for t in tools) - - def test_memory_false_returns_retrieval_only(self): - tools = ocg_tools(url=URL, memory=False) - assert len(tools) == 3 - assert {t.name for t in tools} == { - "ocg_query", - "ocg_get_entity", - "ocg_neighborhood", - } - - def test_subset_switches(self): - tools = ocg_tools(url=URL, entities=False, memory=False) - assert [t.name for t in tools] == ["ocg_query"] - - def test_url_is_required(self): - # There is no server-side default instance — every OCG tool set - # binds its own. - with pytest.raises(TypeError): - ocg_tools() - with pytest.raises(ValueError, match="url"): - ocg_tools(url="") - - def test_instance_binding_lands_in_config(self): - tools = ocg_tools(url=URL, credential="OCG_US_KEY") - for t in tools: - assert t.config["url"] == URL - # Auth rides a standard http-tool header placeholder; the server - # resolves ${NAME} from the credential store at execution. - assert t.config["headers"] == {"Authorization": "Bearer ${OCG_US_KEY}"} - # Declared so the execution token bounds credential resolution - # (same wire contract as http_tool headers). - assert t.credentials == ["OCG_US_KEY"] - - def test_endpoint_mapping(self): - by_name = {t.name: t for t in ocg_tools(url=URL)} - q = by_name["ocg_query"].config - assert (q["method"], q["pathTemplate"]) == ("POST", "/api/v1/agent/query") - e = by_name["ocg_get_entity"].config - assert (e["method"], e["pathTemplate"]) == ("GET", "/api/v1/entities/{entity_id}") - n = by_name["ocg_neighborhood"].config - assert n["pathTemplate"] == "/api/v1/graph/neighborhood/{entity_id}" - assert n["queryParams"] == ["depth", "limit"] - r = by_name["ocg_memory_reinforce"].config - assert (r["method"], r["pathTemplate"]) == ("POST", "/api/v1/memories/{key}/reinforce") - d = by_name["ocg_memory_delete"].config - assert (d["method"], d["pathTemplate"]) == ("DELETE", "/api/v1/memories/{key}") - assert d["queryParams"] == ["agent", "user"] - - def test_trailing_slash_stripped_from_url(self): - tools = ocg_tools(url="https://us.ocg.example.com/") - assert all(t.config["url"] == "https://us.ocg.example.com" for t in tools) - - def test_url_without_credential_is_allowed(self): - tools = ocg_tools(url="https://local-ocg:8080") - for t in tools: - assert t.config["url"] == "https://local-ocg:8080" - assert "headers" not in t.config - assert t.credentials == [] - - def test_credential_without_url_raises(self): - with pytest.raises(TypeError): - ocg_tools(credential="OCG_US_KEY") - - def test_schemas_have_required_fields(self): - by_type = {t.name: t for t in ocg_tools(url=URL)} - assert by_type["ocg_query"].input_schema["required"] == ["query"] - assert by_type["ocg_get_entity"].input_schema["required"] == ["entity_id"] - # LLM-visible hard ceiling on result-set size. - assert by_type["ocg_query"].input_schema["properties"]["max_results"]["maximum"] == 100 - assert by_type["ocg_memory_set"].input_schema["required"] == [ - "key", - "agent", - "user", - "string_value", - "description", - ] - - -class TestOcgAgent: - def test_returns_plain_agent(self): - from conductor.ai.agents.agent import Agent - - agent = ocg_agent(model="anthropic/claude-sonnet-4-6", url=URL) - assert isinstance(agent, Agent) - assert agent.name == "ocg_agent" - assert agent.model == "anthropic/claude-sonnet-4-6" - assert agent.max_turns == 10 - - def test_model_is_required(self): - with pytest.raises(TypeError): - ocg_agent(url=URL) # no model - - def test_url_is_required(self): - with pytest.raises(TypeError): - ocg_agent(model="anthropic/claude-sonnet-4-6") # no url - - def test_canned_prompt_is_default(self): - agent = ocg_agent(model="anthropic/claude-sonnet-4-6", url=URL) - assert agent.instructions == OCG_SYSTEM_PROMPT - # Execution-time date anchor must survive into the prompt verbatim — - # Conductor substitutes it when the LLM task is scheduled. - assert "${workflow.input.__today__}" in agent.instructions - - def test_instructions_override(self): - agent = ocg_agent( - model="anthropic/claude-sonnet-4-6", url=URL, instructions="Custom retrieval prompt." - ) - assert agent.instructions == "Custom retrieval prompt." - - def test_instance_binding_flows_to_tools(self): - agent = ocg_agent( - name="ocg_us", - model="anthropic/claude-sonnet-4-6", - url="https://us.ocg.example.com", - credential="OCG_US_KEY", - ) - assert agent.name == "ocg_us" - from conductor.ai.agents.tool import get_tool_def - - tool_defs = [get_tool_def(t) for t in agent.tools] - assert len(tool_defs) == 6 - for td in tool_defs: - assert td.config["url"] == "https://us.ocg.example.com" - assert td.config["headers"] == {"Authorization": "Bearer ${OCG_US_KEY}"} - assert td.credentials == ["OCG_US_KEY"] - - def test_tool_subset_flags_forwarded(self): - agent = ocg_agent(model="anthropic/claude-sonnet-4-6", url=URL, memory=False) - assert len(agent.tools) == 3 - - def test_exported_from_agents_package(self): - from conductor.ai.agents import ocg_agent as exported_agent - from conductor.ai.agents import ocg_tools as exported_tools - - assert exported_agent is ocg_agent - assert exported_tools is ocg_tools - - -class TestOcgWireFormat: - def test_serializes_with_instance_config(self): - """The serialized agent_tool child must carry each OCG tool's - toolType + config so ToolCompiler can bake the instance binding.""" - from conductor.ai.agents.agent import Agent - from conductor.ai.agents.config_serializer import AgentConfigSerializer - from conductor.ai.agents.tool import agent_tool - - retriever = ocg_agent( - name="ocg_us", - model="anthropic/claude-sonnet-4-6", - url="https://us.ocg.example.com", - credential="OCG_US_KEY", - ) - main = Agent( - name="main", - model="openai/gpt-4o", - instructions="Delegate retrieval.", - tools=[agent_tool(retriever)], - ) - - serialized = AgentConfigSerializer().serialize(main) - - at = serialized["tools"][0] - assert at["toolType"] == "agent_tool" - child = at["config"]["agentConfig"] - assert child["name"] == "ocg_us" - ocg_query = next(t for t in child["tools"] if t["name"] == "ocg_query") - # OCG tools are plain http tools on the wire. - assert ocg_query["toolType"] == "http" - assert ocg_query["config"]["url"] == "https://us.ocg.example.com" - assert ocg_query["config"]["pathTemplate"] == "/api/v1/agent/query" - assert ocg_query["config"]["headers"] == {"Authorization": "Bearer ${OCG_US_KEY}"} - assert ocg_query["config"]["credentials"] == ["OCG_US_KEY"] diff --git a/tests/unit/ai/test_ocg_config.py b/tests/unit/ai/test_ocg_config.py new file mode 100644 index 00000000..35fa46e8 --- /dev/null +++ b/tests/unit/ai/test_ocg_config.py @@ -0,0 +1,98 @@ +"""Tests for declarative OCG configuration.""" + +import pytest + +from conductor.ai.agents import Agent, OcgConfig +from conductor.ai.agents.config_serializer import AgentConfigSerializer + + +def serialize(agent: Agent) -> dict: + return AgentConfigSerializer().serialize(agent) + + +def test_ocg_absent_omits_long_term_memory(): + config = serialize(Agent(name="assistant", model="openai/gpt-4o")) + + assert "longTermMemory" not in config + + +def test_ocg_uses_default_credential_and_generated_agent_identity(): + config = serialize( + Agent( + name="assistant", + model="openai/gpt-4o", + ocg=OcgConfig(url="https://ocg.example.com"), + ) + ) + + assert config["longTermMemory"] == { + "ocgUrl": "https://ocg.example.com", + "credential": "OCG_PUBLIC_KEY", + "agent": "conductor-agent:assistant", + } + + +def test_ocg_serializes_explicit_credential_and_user(): + config = serialize( + Agent( + name="assistant", + model="openai/gpt-4o", + ocg=OcgConfig( + url="https://ocg.example.com", + credential="ASSISTANT_OCG_KEY", + user="user:123", + ), + ) + ) + + assert config["longTermMemory"] == { + "ocgUrl": "https://ocg.example.com", + "credential": "ASSISTANT_OCG_KEY", + "agent": "conductor-agent:assistant", + "user": "user:123", + } + + +@pytest.mark.parametrize( + ("raw_url", "normalized_url"), + [ + (" https://ocg.example.com/ ", "https://ocg.example.com"), + ("https://ocg.example.com///", "https://ocg.example.com"), + ], +) +def test_ocg_normalizes_url(raw_url, normalized_url): + ocg = OcgConfig(url=raw_url) + + assert ocg.url == normalized_url + + +@pytest.mark.parametrize("url", ["", " ", "///"]) +def test_ocg_rejects_empty_url(url): + with pytest.raises(ValueError, match="url must be non-empty"): + OcgConfig(url=url) + + +@pytest.mark.parametrize("credential", ["", " "]) +def test_ocg_rejects_empty_credential(credential): + with pytest.raises(ValueError, match="credential must be a non-empty secret name"): + OcgConfig(url="https://ocg.example.com", credential=credential) + + +def test_ocg_does_not_accept_raw_api_key_parameter(): + with pytest.raises(TypeError): + OcgConfig(url="https://ocg.example.com", api_key="raw-secret") + + +def test_ocg_emits_no_legacy_feedback_worker_or_config(): + config = serialize( + Agent( + name="assistant", + model="openai/gpt-4o", + ocg=OcgConfig(url="https://ocg.example.com"), + ) + ) + + assert "feedbackSink" not in config + assert "feedbackWorker" not in config + assert "memorySummaryModel" not in config["longTermMemory"] + assert "tools" not in config