simon824 commented on code in PR #380: URL: https://github.com/apache/hugegraph-ai/pull/380#discussion_r4336346491
########## server/ontogeny/mcp_server.py: ########## @@ -0,0 +1,486 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""MCP server: the ontology as an agent tool surface. + +Two transports, ONE door. External MCP clients (Claude Desktop, ZCode, +LangGraph, ...) connect over streamable HTTP at ``/mcp``; the builtin LLM +engine drives the same tool surface over an in-memory MCP session. Both +funnel through :meth:`AgentBroker.call_tool` when a session is given, so +budget, catalog, approval gates and the step trail (I4) are IDENTICAL to +the REST session protocol -- MCP is a transport, not a second door. + +Call shapes (every catalog tool accepts these reserved arguments): + +- ``plugin`` + ``session_id``: execute as the plugin's frozen identity + inside that session -- budget charged, writes gated, steps recorded. +- no session: read-only exploration. Reads run under the plugin principal + narrowed to the plugin's own tool list (or the legacy shared + ``mcp-agent`` identity, which role-gated policy refuses by design); + writes are refused -- a session is the governed execution container + (agent-paradigm §1). +- ``thought``: the caller's own reasoning for this step, recorded on the + trail for human review (I4); never part of tool parameters. + +Session-management tools (``agent_open_session`` ...) let a pure MCP client +run the whole lifecycle; approval DECISIONS stay with the human console. + +Requires the optional ``mcp`` extra (mcp>=2.2); absent, the surface simply +does not mount and the builtin engine falls back to in-process calls. +""" +from __future__ import annotations + +import asyncio +import inspect +import json +import logging +import os +from contextlib import asynccontextmanager +from contextvars import ContextVar +from typing import Any, AsyncIterator, Iterable + +from .errors import OOError +from .service import ServiceContext + +log = logging.getLogger("ontogeny.mcp") + +MCP_HTTP_PATH = "/mcp" +#: Comma-separated host allowlist; when set, the SDK's DNS-rebinding +#: protection is enabled for exactly these hosts. Unset (default) matches +#: the REST API's posture: no Host-header policing. +_ALLOWED_HOSTS_ENV = "ONTOGENY_MCP_ALLOWED_HOSTS" + +#: The authenticated driver behind the current /mcp request (set by the +#: bearer gate middleware in ontogeny.api.app); tools read it so the session +#: trail records WHO drove, not just that the MCP transport did. +_MCP_DRIVER: ContextVar[dict[str, Any] | None] = ContextVar("ontogeny_mcp_driver", default=None) + + +def set_mcp_driver(principal: dict[str, Any] | None) -> None: + _MCP_DRIVER.set(principal) + + +def current_mcp_driver() -> dict[str, Any] | None: + return _MCP_DRIVER.get() + + +#: Header that scopes an MCP client's tools/list to ONE plugin's effective +#: catalog (plus the session tools). Large ontologies compile hundreds of +#: search_* tools; a plugin-scoped client (or an LLM behind it) sees only +#: what its plugin may call. Purely a VIEW filter -- call_tool enforcement +#: is unchanged (broker + frozen catalog). +PLUGIN_SCOPE_HEADER = "x-ontogeny-plugin" + +_SESSION_TOOLS = {"agent_open_session", "agent_get_session", + "agent_list_sessions", "agent_finish_session"} + + +class PluginScopeMiddleware: + """tools/list through a plugin-colored lens (see PLUGIN_SCOPE_HEADER). + + Runs at the context tier where the streamable-HTTP transport has attached + the raw request; without the header (in-memory engine traffic, plain + clients) it is a no-op pass-through. + """ + + def __init__(self, sc: ServiceContext) -> None: + self.sc = sc + + async def __call__(self, ctx, call_next): + result = await call_next(ctx) + if ctx.method != "tools/list": + return result + plugin = None + req = getattr(ctx, "request", None) + getter = getattr(req, "headers", None) + if getter is not None: + try: + plugin = getter.get(PLUGIN_SCOPE_HEADER) + except Exception: # noqa: BLE001 -- transport-specific header obj + plugin = None + if not plugin: + return result + compiled = self.sc.compiled + plug = compiled.agent_plugins.get(plugin) if compiled else None + if plug is None: + from mcp.shared.exceptions import MCPError + + raise MCPError(-32000, f"unknown plugin {plugin!r} in {PLUGIN_SCOPE_HEADER}") + from .agent.catalog import effective_tools + + keep = set(effective_tools(compiled, plug)) | _SESSION_TOOLS + + def _name(t): + return t.name if hasattr(t, "name") else t.get("name") + + # the middleware runs pre-serialization: some SDK paths hand the + # pydantic result, others the wire dict -- filter both shapes + if hasattr(result, "tools"): + result.tools = [t for t in result.tools if _name(t) in keep] + elif isinstance(result, dict) and "tools" in result: + result["tools"] = [t for t in result["tools"] if _name(t) in keep] + return result + + +def mcp_sdk() -> Any: + """The ``mcp`` 2.x SDK module, or None when the optional extra is absent.""" + try: + import mcp.server.mcpserver as mod + except ImportError: + return None + return mod + + +def build_mcp_server(sc: ServiceContext, *, engine_mode: bool = False): + """Compatibility shim: one server instance, tools built from the current + compiled snapshot. Returns ``None`` when the extra is missing.""" + surface = McpSurface(sc, engine_mode=engine_mode) + surface.build() + return surface.server + + +def _first_text(content: Iterable[Any]) -> str: + for block in content or []: + text = getattr(block, "text", None) + if text: + return text + return "" + + +# --------------------------------------------------------------------- client + + +class McpToolClient: + """A thin envelope decoder over an MCP ``ClientSession``. + + The engine (and tests) use this to speak the MCP protocol without + caring about wire details: tools return the broker's result envelope + verbatim as structured content; wire/protocol failures become outcome + envelopes instead of exceptions, matching the session protocol's + structured-outcome contract (agent-paradigm §3.1). + """ + + def __init__(self, session: Any) -> None: + self.session = session + + async def list_tools(self) -> list[dict[str, Any]]: + res = await self.session.list_tools() + return [{"name": t.name, "description": t.description or "", + "input_schema": getattr(t, "input_schema", None) or getattr(t, "inputSchema", {}) + or {}} for t in res.tools] + + async def call_tool(self, name: str, args: dict[str, Any]) -> dict[str, Any]: + from mcp.shared.exceptions import MCPError + + try: + res = await self.session.call_tool(name, args) + except MCPError as exc: # protocol-level refusal (unknown tool, ...) + text = str(exc) + if "unknown tool" in text.lower(): + return {"outcome": "NOT_FOUND", "error": text} + return {"outcome": "MCP_TOOL_ERROR", "error": text} + if getattr(res, "is_error", False): + text = _first_text(res.content) + if text.lower().startswith("unknown tool"): + return {"outcome": "NOT_FOUND", "error": text} + return {"outcome": "MCP_TOOL_ERROR", "error": text} + structured = getattr(res, "structured_content", None) + if isinstance(structured, dict): + return structured + text = _first_text(res.content) + try: + decoded = json.loads(text) if text else {} + except (TypeError, ValueError): + return {"outcome": "ok", "detail": text} + return decoded if isinstance(decoded, dict) else {"outcome": "ok", "detail": decoded} + + +# --------------------------------------------------------------------- surface + + +class McpSurface: + """The MCP tool surface over one ServiceContext. + + ``engine_mode=True`` builds the instance the builtin engine drives: + broker calls carry ``engine=True`` so the driving engine may own a + ``running`` session (agent-paradigm §2). That instance is reachable + only over in-process memory streams, never over the wire. + """ + + def __init__(self, sc: ServiceContext, *, engine_mode: bool = False) -> None: + self.sc = sc + self.engine_mode = engine_mode + self.sdk = mcp_sdk() + self.server: Any = None + self._mounted = False + + # -------------------------------------------------------------- building + + def build(self) -> None: + """(Re)build the tool registrations from the CURRENT compiled snapshot. + + Called at mount time and again whenever the ontology is promoted -- + the tool list is a compiled artifact, never hand-maintained. + """ + if self.sdk is None or self.sc.compiled is None: + return + if self.server is None: + kwargs: dict[str, Any] = {} + if not self.engine_mode: + # the wire surface gets the plugin-scope lens; the engine's + # private surface drives with the session's frozen catalog + kwargs["middleware"] = [PluginScopeMiddleware(self.sc)] + self.server = self.sdk.MCPServer( + "ontogeny", + title="Ontogeny — operational ontology", + description="Operational ontology as an agent tool surface", + instructions=( + "Tools mirror the governed session protocol. Pass plugin " + "and session_id (see agent_open_session) to run inside a " + "session's budget, catalog and approval gates; without a " + "session calls are read-only. Send the " + f"'{PLUGIN_SCOPE_HEADER}: <plugin>' header on tools/list to " + "see only one plugin's effective catalog." + ), + **kwargs, + ) + else: + for tool in list(self.server._tool_manager.list_tools()): + self.server.remove_tool(tool.name) + self._register_catalog_tools() + self._register_session_tools() + + def refresh(self) -> None: + """Rebuild tool registrations after the compiled snapshot changed.""" + if self.server is not None: + self.build() + + def _register_catalog_tools(self) -> None: + from .agent.catalog import build_catalog + + for name, tdef in sorted(build_catalog(self.sc.compiled).items()): + self.server.add_tool( + _catalog_tool_fn(self.sc, name, tdef, self.engine_mode), + name=name, + description=tdef.description, + structured_output=True, + ) + + def _register_session_tools(self) -> None: + sc, server = self.sc, self.server + + async def agent_open_session(plugin: str, task: str = "") -> dict[str, Any]: + """Open an agent session for a declared AgentPlugin; returns the + session view incl. the tools you may call and the id to pass as + session_id on every tool call.""" + return await sc.agents.open_session(plugin, task or "", + driver=current_mcp_driver()) + + async def agent_get_session(session_id: int) -> dict[str, Any]: + """Session view: status, budget usage, effective tools, full step + trail (thought + outcome per step) -- the driver's transcript.""" + return await sc.agents.get_session(session_id) + + async def agent_list_sessions(plugin: str | None = None, + limit: int = 50) -> dict[str, Any]: + """List agent sessions, newest first (optionally per plugin).""" + return {"sessions": await sc.agents.list_sessions(plugin=plugin, + limit=min(int(limit), 200))} + + async def agent_finish_session(session_id: int, + result: dict | None = None) -> dict[str, Any]: + """Finish a session with a final result (the human-facing act).""" + return await sc.agents.finish_session(session_id, result) + + for fn in (agent_open_session, agent_get_session, + agent_list_sessions, agent_finish_session): + server.add_tool(fn, structured_output=True) + + # --------------------------------------------------------------- serving + + def graft(self, app: Any) -> bool: + """Attach the streamable-HTTP route(s) to a FastAPI/Starlette app. + + Grafting the routes directly (instead of ``Mount``) serves exactly + ``/mcp`` with no 307 redirect games; the SPA fallback already + excludes the path, and MCP clients only ever POST in JSON mode. + Idempotent per app. + """ + if self.sdk is None or self.server is None: + return False + if self._mounted: + return True + kwargs: dict[str, Any] = dict( + streamable_http_path=MCP_HTTP_PATH, + # POST + JSON responses, one request per call: survives proxies + # and test ASGI transports, and matches the stateless call shape + # (every call carries plugin/session_id anyway) + json_response=True, + stateless_http=True, + ) + # the SDK auto-enables DNS-rebinding protection for localhost hosts; + # ontogeny's REST surface does no Host policing, so the MCP surface + # matches it by default and ONTOGENY_MCP_ALLOWED_HOSTS opts INTO a list + from mcp.server.transport_security import TransportSecuritySettings + hosts = [h.strip() for h in os.environ.get(_ALLOWED_HOSTS_ENV, "").split(",") if h.strip()] + kwargs["transport_security"] = ( + TransportSecuritySettings(enable_dns_rebinding_protection=True, allowed_hosts=hosts) + if hosts else + TransportSecuritySettings(enable_dns_rebinding_protection=False) + ) + starlette = self.server.streamable_http_app(**kwargs) + existing = {getattr(route, "path", None) for route in app.router.routes} + for route in starlette.routes: + if getattr(route, "path", None) not in existing: + app.router.routes.append(route) + self._mounted = True + return True + + @asynccontextmanager + async def run(self, app: Any) -> AsyncIterator[Any]: + """Serve: build (if needed), graft, run the HTTP session manager.""" + if self.sdk is None: + log.info("MCP surface disabled: the optional extra is not installed " + "(pip install 'ontogeny[mcp]')") + yield None + return + if self.server is None: + self.build() + if self.server is None: # compiled not ready yet + yield None + return + self.graft(app) + async with self.server.session_manager.run(): + log.info("MCP surface serving at %s (json, stateless)", MCP_HTTP_PATH) + yield self.server + + +@asynccontextmanager +async def mount_mcp(app: Any, sc: ServiceContext) -> AsyncIterator[Any]: + """Mount + run the external MCP surface on a FastAPI app (lifespan-side). + + The surface is stored on ``sc`` so a later promote can refresh it + (``sc.mcp_surface.refresh()``). + """ + surface = McpSurface(sc) + sc.mcp_surface = surface # type: ignore[attr-defined] + async with surface.run(app): + yield surface + + +@asynccontextmanager +async def in_process_mcp_session(sc: ServiceContext, *, + engine_mode: bool = True) -> AsyncIterator[McpToolClient]: + """The builtin engine's transport: a real MCP client session over + in-memory streams against a private server. + + ``engine_mode=True`` (the default, used by the engine) lets the driving + engine own a ``running`` session; ``False`` is the in-process stand-in + for an EXTERNAL client -- tests use it to prove the §2 isolation the + wire surface enforces. + + Same protocol, same handlers, same governance as the wire surface -- + the reference engine dogfoods the surface external agents use, without + sockets (tests stay hermetic). Cancellation-safe: the server task is + always collected. + """ + from mcp.client.session import ClientSession + from mcp.shared.memory import create_client_server_memory_streams + + surface = McpSurface(sc, engine_mode=engine_mode) + surface.build() + if surface.server is None: + raise OOError("engine MCP session requires the optional dependency: " + "pip install 'ontogeny[mcp]' (and a compiled ontology)", + details={"missing": "mcp"}) + low = surface.server._lowlevel_server + async with create_client_server_memory_streams() as (client_side, server_side): + server_task = asyncio.create_task( + low.run(server_side[0], server_side[1], low.create_initialization_options())) + try: + async with ClientSession(client_side[0], client_side[1]) as sess: + await sess.initialize() + yield McpToolClient(sess) + finally: + server_task.cancel() + try: + await server_task + except BaseException: # noqa: BLE001 -- cancelled server task + pass + + +# ------------------------------------------------------------ tool generation + +#: catalog input_schema type -> signature annotation. Anything object/array +#: shaped maps to Any: the schema stays permissive, the broker validates. +_SCHEMA_ANN: dict[str, Any] = { + "integer": int | None, "string": str | None, + "number": float | None, "boolean": bool | None, +} + + +def _catalog_tool_fn(sc: ServiceContext, name: str, tdef, engine_mode: bool): + """One MCP tool = one catalog ToolDef, dispatched through the broker + when a session is given (the governed path), else read-only direct. + + The tool's signature is generated from the ToolDef's input_schema so + clients (and the LLM behind them) see the same parameter shape as the + REST session protocol's ``input_schema``. + """ + props: dict[str, Any] = (tdef.input_schema or {}).get("properties") or {} + + async def tool(plugin: str | None = None, session_id: int | None = None, + thought: str | None = None, **kwargs: Any) -> dict[str, Any]: + if session_id is not None: + return await sc.agents.call_tool(int(session_id), name, kwargs, Review Comment: Fixed (commit a39a838): MCP session tool calls now pass `driver=_MCP_DRIVER.get()` so steps record WHO drove, not just the plugin identity. ########## ontogeny/server/ontogeny/api/routes_evolve.py: ########## @@ -0,0 +1,231 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""The RSI loop's console surface (admin-gated at the router).""" +from __future__ import annotations + +from pathlib import Path + +from fastapi import APIRouter, Depends, HTTPException +from sqlalchemy import select + +from ..evolve.models import EvolveSignalRow, ProposalRow +from ..evolve.ops import _evolve_impact, run_diagnose +from ..evolve.promoter import apply_mutations +from ..core.loader import load_package +from ..service import ServiceContext +from .deps import get_sc +from .schemas import GraduationBody, RejectBody + + +def register(evolve_r: APIRouter, sc: ServiceContext) -> None: + # ---- evolve (RSI loop) ------------------------------------------------- + + + @evolve_r.get("/signals") + async def signals(limit: int = 100, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + rows = (await s.execute( + select(EvolveSignalRow).order_by(EvolveSignalRow.id.desc()).limit(limit) + )).scalars().all() + return {"signals": [{"id": x.id, "kind": x.kind, "evidence": x.evidence} for x in rows]} + + @evolve_r.delete("/signals/{signal_id}") + async def delete_signal(signal_id: int, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + row = (await s.execute(select(EvolveSignalRow).where(EvolveSignalRow.id == signal_id))).scalar_one_or_none() + if row is None: + raise HTTPException(404, "signal not found") + await s.delete(row) + await s.commit() + return {"deleted": signal_id} + + @evolve_r.post("/cycle") + async def evolve_cycle(sc: ServiceContext = Depends(get_sc)): + """One scheduled beat: aggregate signals, then diagnose (direction F). + Promotion is NEVER automatic from here -- the ladder keeps its gates.""" + async with sc.sessionmaker() as s: + evo = getattr(sc.compiled, "evolution", None) + obs = getattr(getattr(evo, "spec", None), "observability", None) if evo else None + created = await sc.telemetry.aggregate_signals( + s, thresholds=getattr(obs, "thresholds", None) or None) + await s.commit() + diagnosed = await run_diagnose(sc) + return {"signals_created": created, **diagnosed} + + @evolve_r.post("/aggregate") + async def aggregate_signals(sc: ServiceContext = Depends(get_sc)): + # aggregate_signals runs every detector (telemetry + quarantine) and is + # idempotent within the window: an unchanged world yields zero new rows + async with sc.sessionmaker() as s: + # per-domain sensitivity from EvolutionPolicy.observability.thresholds + evo = getattr(sc.compiled, "evolution", None) + thresholds = getattr(getattr(evo, "spec", None), "observability", None) if evo else None + created = await sc.telemetry.aggregate_signals( + s, thresholds=getattr(thresholds, "thresholds", None) or None) + await s.commit() + return {"created": created} + + @evolve_r.post("/diagnose") + async def diagnose_signals(sc: ServiceContext = Depends(get_sc)): + """Signals -> gaps -> proposals (see run_diagnose).""" + return await run_diagnose(sc) + + @evolve_r.get("/proposals") + async def list_proposals(status: str | None = None, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + stmt = select(ProposalRow).order_by(ProposalRow.id.desc()).limit(100) + if status: + stmt = stmt.where(ProposalRow.status == status) + rows = (await s.execute(stmt)).scalars().all() + return {"proposals": [{"id": p.id, "status": p.status, "tier": p.tier, "gap_kind": p.gap_kind, + "origin": p.origin, "llm_analysis": p.llm_analysis, + "superseded_by": p.superseded_by, + "impact": _evolve_impact(sc.compiled, p.diff)} for p in rows]} + + @evolve_r.get("/proposals/{proposal_id}") + async def get_proposal(proposal_id: int, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + p = (await s.execute(select(ProposalRow).where(ProposalRow.id == proposal_id))).scalar_one_or_none() + if p is None: + raise HTTPException(404, "proposal not found") + return {"id": p.id, "status": p.status, "tier": p.tier, "origin": p.origin, "diff": p.diff, + "rationale": p.rationale, "eval_report": p.eval_report, + "llm_analysis": p.llm_analysis, "rejected_reason": p.rejected_reason, + "superseded_by": p.superseded_by, "signal_id": p.signal_id, + "impact": _evolve_impact(sc.compiled, p.diff)} + + @evolve_r.delete("/proposals/{proposal_id}") + async def delete_proposal(proposal_id: int, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + row = (await s.execute(select(ProposalRow).where(ProposalRow.id == proposal_id))).scalar_one_or_none() + if row is None: + raise HTTPException(404, "proposal not found") + await s.delete(row) + await s.commit() + return {"deleted": proposal_id} + + @evolve_r.post("/proposals/{proposal_id}/reject") + async def reject_proposal(proposal_id: int, body: RejectBody, + sc: ServiceContext = Depends(get_sc)): + """The human 'no' is a first-class outcome AND the loop's memory: the + reason is fed back into the proposer's prompt on later rounds, so the + same rejected mutation is not re-filed verbatim (direction E).""" + async with sc.sessionmaker() as s: + row = (await s.execute(select(ProposalRow).where(ProposalRow.id == proposal_id))).scalar_one_or_none() + if row is None: + raise HTTPException(404, "proposal not found") + row.status = "rejected" + row.rejected_reason = (body.reason or "").strip() or None + await s.commit() + return {"id": proposal_id, "status": "rejected"} + + + + + @evolve_r.get("/graduation") + async def graduation_status(sc: ServiceContext = Depends(get_sc)): + """Direction B3: how close each mutation class is to earning more + autonomy (T1/T2 -> T0). Earning it stays a T3 human decision.""" + from ..evolve.promoter import GRADUATION_STREAK, graduation_streak + from ..evolve.proposer import MUTATION_CATALOG + + evo = getattr(sc.compiled, "evolution", None) + tiers = getattr(getattr(evo, "spec", None), "tiers", {}) or {} + t0 = set(getattr(tiers.get("t0-auto-merge"), "mutations", []) or []) + async with sc.sessionmaker() as s: + classes = [] + for kind in MUTATION_CATALOG: + if kind in t0: + continue # already trusted at T0: nothing to graduate + streak = await graduation_streak(s, kind) + classes.append({"mutation": kind, "streak": streak, + "threshold": GRADUATION_STREAK, + "eligible": streak >= GRADUATION_STREAK}) + return {"classes": classes, + "rule": getattr(getattr(getattr(evo, "spec", None), "graduation", None), "rule", None)} + + @evolve_r.post("/graduation/request") + async def graduation_request(body: GraduationBody, sc: ServiceContext = Depends(get_sc)): + """File the T3 ticket: promote a mutation class toward T0. The request + is created for human execution only -- EvolutionPolicy is a constitution + file the loop never edits itself.""" + from ..evolve.promoter import GRADUATION_STREAK, graduation_streak + from ..evolve.proposer import MUTATION_CATALOG + + if body.mutation not in MUTATION_CATALOG: + raise HTTPException(400, f"unknown mutation class {body.mutation!r}") + evo = getattr(sc.compiled, "evolution", None) + tiers = getattr(getattr(evo, "spec", None), "tiers", {}) or {} + t0 = set(getattr(tiers.get("t0-auto-merge"), "mutations", []) or []) + if body.mutation in t0: + raise HTTPException(400, f"{body.mutation!r} is already T0") + async with sc.sessionmaker() as s: + streak = await graduation_streak(s, body.mutation) + if streak < GRADUATION_STREAK: + raise HTTPException(429, f"streak {streak}/{GRADUATION_STREAK} not enough to graduate") + row = ProposalRow(gap_kind="graduation-request", + diff=[{"mutation": "tier-change", "mutation_class": body.mutation, + "to": "t0-auto-merge"}], + rationale=(f"mutation class {body.mutation!r} has {streak} consecutive " + f"clean promotions; request to graduate to T0 (human decision)"), + origin="human", status="awaiting_human") + s.add(row) + await s.commit() + await s.flush() + return {"id": row.id, "status": row.status, "streak": streak} + + @evolve_r.post("/proposals/{proposal_id}/eval") + async def eval_proposal(proposal_id: int, sc: ServiceContext = Depends(get_sc)): + async with sc.sessionmaker() as s: + p = (await s.execute(select(ProposalRow).where(ProposalRow.id == proposal_id))).scalar_one_or_none() + if p is None: + raise HTTPException(404, "proposal not found") + # B1: the ladder must judge the CANDIDATE, not the live schema -- + # apply the diff to an in-memory copy and let the runner see the + # candidate's declared columns and (crucially) its action rules + # during time-travel replay. A diff no applier can express (manual + # T1/T2 kinds) still evaluates against live, honestly labelled. + candidate = None + candidate_error = None + if sc.package_root is not None: + try: + candidate = apply_mutations(load_package(Path(sc.package_root)), p.diff) + except Exception as exc: # noqa: BLE001 -- unappliable diff is a report, not a crash + candidate_error = f"{type(exc).__name__}: {exc}" + reports = {} + passed = True Review Comment: Fixed (commit 9cfda13): `passed` starts `False` when the package declares zero eval suites — a vacuous green is no longer possible. ########## ontogeny/server/ontogeny/service.py: ########## @@ -0,0 +1,1626 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""ServiceContext: the single composition root. + +REST API, MCP server, CLI and background workers are thin shells over this +object -- governance (transactions, policy, audit) has exactly one path. +""" +from __future__ import annotations + +import logging +import os +import re +from dataclasses import replace +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +import httpx + +from .action import ActionRuntime, OutboxDispatcher, SseBroker +from .auth import AuthService +from .agent.broker import AgentBroker +from .config import Settings +from .core.loader import load_package +from .db import init_schema, make_engine, make_sessionmaker +from .engine import QueryService +from .errors import DSLValidationError, NotFoundError, OOError, StoreError +from .evolve import Promoter +from .evolve.evalrunner import EvalRunner +from .ext_host import ExtensionHost +from .functions import DeclarativeRuntime, DerivationWorker, FunctionRuntime, FunctionSandbox +from .policy import PolicyEngine +from .projection import ProjectionWorker, compile_projection +from .registry import RegistryService +from .registry.compiled import CompiledOntology, domain_graph_name +from .stores import ObjectRepository, SyncEngine +from .stores.sources import make_source +from .telemetry import TelemetryService + +log = logging.getLogger("ontogeny.service") + +_DOMAIN_NAME_RE = re.compile(r"[a-z][a-z0-9-]{0,62}") +# resource directories a scaffolded (empty) domain ships with, so the builder +# and hand-editing both have the conventional layout to write into +_DOMAIN_DIRS = ("objects", "links", "actions", "functions", "policies", "stores", "evals") + + +class DomainError(OOError): + """Bad domain name or the domain already exists.""" + + code = "DOMAIN_INVALID" + http_status = 400 + + +def _read_manifest(path: Path) -> dict[str, Any]: + """ontology.yaml of the package at ``path`` as a plain dict ({} on any + failure -- listing must survive a half-written domain directory).""" + try: + import yaml + + doc = yaml.safe_load((Path(path) / "ontology.yaml").read_text(encoding="utf-8")) + return doc if isinstance(doc, dict) else {} + except Exception: # noqa: BLE001 -- listing is best-effort by design + return {} + + +def _manifest_name(path: Path) -> str | None: + meta = _read_manifest(path).get("metadata") or {} + name = meta.get("name") + return str(name) if isinstance(name, str) else None + + +def _count_yaml(path: Path) -> int: + return len(list(path.glob("*.yaml"))) if path.is_dir() else 0 + + +def _check_package(path: Path) -> dict[str, Any]: + """Dynamic health check for one domain: load the package from disk and run + the semantic validator, the same bar activation will apply. Listing stays + best-effort: a package that fails to parse reports ``ok: false`` with the + first errors instead of breaking the switcher.""" + try: + from .core import load_package, validate + + rep = validate(load_package(str(path))) + if rep.ok: + return {"ok": True, "errors": []} + return {"ok": False, + "errors": [f"{i.resource}: {i.message}" for i in rep.issues if i.severity == "error"][:5]} + except Exception as exc: # noqa: BLE001 -- any failure is the check's answer + return {"ok": False, "errors": [str(exc)[:300]]} + + +def _pk_from_graph_id(graph_id: Any) -> str | None: + """HugeGraph stores a projected object as ``<labelId>:<primaryKey>``.""" + if graph_id is None: + return None + text = str(graph_id) + key = text.split(":", 1)[1] if ":" in text else text + return key or None + + +#: Why the projection layer is not wired, in one sentence each. The console +#: shows these instead of a generic failure, because each one has a DIFFERENT +#: fix and none of them is "declare a projection". +_PROJECTION_BLOCK_REASONS = { + "not-declared": "no projection declared", + "provider-sqlite": "storage provider is {provider}, not 'hugegraph', so nothing is projected", + "endpoint-unresolved": "the projection endpoint ${HUGEGRAPH_URL} cannot be resolved " + "(set the HugeGraph address in the storage configuration)", + "extension-missing": "the graph-hugegraph extension is not loaded, so there is no " + "graph client to wire", +} + + +def _display_of(obj: dict, compiled: CompiledOntology, object_type: str) -> str: + """Best human label for a preview row (name-like field, else the key).""" + for candidate in ("name", "display", "title"): + if obj.get(candidate): + return str(obj[candidate]) + pk = compiled.objects[object_type].spec.primaryKey[0] + return str(obj.get(pk, "")) + + +def _mask_dsn(dsn: str) -> str: + """A DSN with the password removed and the noise trimmed, so the console + can show WHERE data lives without leaking credentials.""" + import re as _re + + if not dsn: + return "" + if dsn.startswith(("sqlite", "csv")) or dsn.startswith("/"): + tail = dsn.split("///", 1)[-1] + return f"file: {tail}" if not dsn.startswith("csv") else f"file: {dsn}" + m = _re.match(r"(\w+\+?\w*)://([^:/@]+)(?::[^@]*)?@([^/]+)/(\S+)", dsn) + if m: + return f"{m.group(1)}://{m.group(2)}@{m.group(3)}/{m.group(4)}" + m = _re.match(r"(\w+\+?\w*)://([^/@]+)?@?([^/]+)/(\S+)", dsn) + if m: + user = f"{m.group(2)}@" if m.group(2) else "" + return f"{m.group(1)}://{user}{m.group(3)}/{m.group(4)}" + return dsn[:80] + + +class ServiceContext: + def __init__(self, settings: Settings, package_root: str | None = None, + *, projection_http: httpx.AsyncClient | None = None, + llm_http: httpx.AsyncClient | None = None, + ui_dir: str | None = None, + package_root_override: str | None = None) -> None: + self.settings = settings + self.package_root = package_root + # The package this process booted from (ONTOGENY_PACKAGE_ROOT). Domain + # activation moves `package_root`, but the boot package stays listed + # and switchable -- it is the deployment's default domain. + self.boot_package_root = package_root + # On DB-only boots the snapshot has no directory to point at; if the + # caller knows where the package lives on disk, use it for sidecars + # (function sources, reprojection of policy text). + self.package_root_override = package_root_override + self.ui_dir = ui_dir # built SPA directory served same-origin (optional) + from .layout import LayoutStore + self._layout = LayoutStore(package_root) + self.registry = RegistryService() + self.engine = None + self.sessionmaker = None + self.compiled: CompiledOntology | None = None + self.sse = SseBroker() + self.projection_http = projection_http + self.llm_http = llm_http + # extension-pluggable surfaces (populated by extensions/ at boot): + # sc.llm (capability:llm), sc.graph_store_factory (capability:graph-store), + # sc.agent_engines (engine factories). The core only defines contracts. + self.llm = None + self.graph_store_factory = None + self.agent_engines: dict[str, Any] = {} + self.ext: ExtensionHost | None = None + + # ------------------------------------------------------------- lifecycle + + async def initialize(self) -> "ServiceContext": + self.engine = make_engine(self.settings.db_dsn) + await init_schema(self.engine) + self.sessionmaker = make_sessionmaker(self.engine) + # Persisted console configuration is applied BEFORE extensions load, so + # a saved LLM/HugeGraph endpoint wins over the process environment. + await self._load_runtime_settings() + self.ext = ExtensionHost(self) + self.ext.discover() + self.ext.load() # providers set sc.llm / sc.graph_store_factory; engines register factories + # A deployment with no administrator is unreachable -- nobody could + # create the first account -- so bootstrap runs once, on an empty table, + # right after the schema exists. + self.auth = AuthService(self) + await self.auth.ensure_admin() + async with self.sessionmaker() as session: + compiled = await self.registry.load_latest(session) + if compiled is None: + if self.package_root is None: + raise StoreError("registry is empty and no ONTOGENY_PACKAGE_ROOT configured to publish from") + compiled = await self.publish(self.package_root) + else: + self._rebuild_services(compiled) + async with self.sessionmaker() as session: + await self.repo.ensure(session) + return self + + async def _load_runtime_settings(self) -> None: + """Apply the console-saved configuration row over the process environment. + + "Over" is precise: the row wins where it says something, and stays out of + the way where it does not. A row with no HugeGraph URL is NOT an + instruction to forget ``HUGEGRAPH_URL`` — that variable is usually the + deployment's own (daemon.sh, compose.yaml), and every + ``${HUGEGRAPH_URL}`` in the DSL resolves against it. Deleting it here + turned "the console never stored a URL" into "this deployment has no + graph endpoint", which silently disabled a working projection. Only an + explicit console choice of ``sqlite`` withdraws the endpoint. + """ + from .runtime_config import load_runtime_config + + async with self.sessionmaker() as session: + row = await load_runtime_config(session) + if row is None: + return + env = dict(self.settings.env) + storage_provider = row.storage_provider or ("hugegraph" if row.hugegraph_url else self.settings.storage_provider) + if storage_provider == "hugegraph": + # the row's endpoint, else whatever the process was started with + url = row.hugegraph_url or os.environ.get("HUGEGRAPH_URL") + if url: + env["HUGEGRAPH_URL"] = url + elif row.storage_provider == "sqlite": + env.pop("HUGEGRAPH_URL", None) + os.environ.pop("HUGEGRAPH_URL", None) + self.settings = replace( + self.settings, + llm_provider=row.llm_provider or self.settings.llm_provider or "ollama", + llm_base_url=row.llm_base_url or None, + llm_model=row.llm_model or None, + llm_api_key=row.llm_api_key or None, + storage_provider=storage_provider, + # Credentials follow the same rule as the URL: the row wins where it + # says something, the deployment's own environment otherwise. They + # are deliberately NOT exported into os.environ — an env var is + # world-readable through /proc, and nothing outside this process + # needs them (the graph client is constructed here). + hugegraph_user=row.hugegraph_user or self.settings.hugegraph_user, + hugegraph_password=row.hugegraph_password or self.settings.hugegraph_password, + env=env, + ) + if env.get("HUGEGRAPH_URL"): + os.environ["HUGEGRAPH_URL"] = env["HUGEGRAPH_URL"] + + async def runtime_config(self) -> dict[str, Any]: + """The LLM + storage configuration exposed to the operations console.""" + return { + "llm": { + "provider": self.settings.llm_provider or "ollama", + "base_url": self.settings.llm_base_url, + "model": self.settings.llm_model, + "api_key_set": bool(self.settings.llm_api_key), + }, + "storage": { + "provider": self._storage_provider(), + "url": self.settings.env.get("HUGEGRAPH_URL") or os.environ.get("HUGEGRAPH_URL"), + "user": self.settings.hugegraph_user, + # the password itself never leaves the server, exactly like the + # LLM api key: the console learns only whether one is set + "password_set": bool(self.settings.hugegraph_password), + }, + } + + async def configure_runtime( + self, *, + llm_provider: str | None = None, + llm_base_url: str | None = None, + llm_model: str | None = None, + llm_api_key: str | None = None, + storage_provider: str | None = None, + storage_url: str | None = None, + storage_user: str | None = None, + storage_password: str | None = None, + ) -> dict[str, Any]: + """Persist + hot-apply the LLM and storage provider selections.""" + from .runtime_config import save_runtime_config + + provider = (llm_provider or self.settings.llm_provider or "ollama").lower() + storage = (storage_provider or self.settings.storage_provider or "sqlite").lower() + env = dict(self.settings.env) + # A blank URL field means "keep whatever endpoint is in effect" — the + # deployment's own HUGEGRAPH_URL when the console has none — exactly as a + # blank API key means "keep the secret". It must NOT mean "erase it": + # that silently dropped a working graph endpoint and, because the + # resolution scope is also exported into the process environment, took + # the deployment's own variable down with it. + if storage == "hugegraph": + url = (storage_url or self.settings.env.get("HUGEGRAPH_URL") + or os.environ.get("HUGEGRAPH_URL")) + if url: + env["HUGEGRAPH_URL"] = url + else: + env.pop("HUGEGRAPH_URL", None) + # A blank API key from the form means "keep the existing secret"; the UI + # never receives the key back after it has been saved. + api_key = llm_api_key if llm_api_key is not None else self.settings.llm_api_key + if provider == "disabled": + base_url = None + model = None + api_key = None + else: + base_url = llm_base_url if llm_base_url is not None else self.settings.llm_base_url + model = llm_model if llm_model is not None else self.settings.llm_model + if provider != "external": + api_key = None + # Same convention for the graph credentials: a blank password keeps the + # stored one (the console never receives it back), a blank user clears + # the pair so an operator can drop credentials when a server stops + # enforcing auth. + hg_user = storage_user if storage_user is not None else self.settings.hugegraph_user + if not hg_user: + hg_password = None + hg_user = None + else: + hg_password = (storage_password if storage_password + else self.settings.hugegraph_password) + self.settings = replace( + self.settings, + llm_provider=provider, + llm_base_url=base_url or None, + llm_model=model or None, + llm_api_key=api_key or None, + storage_provider=storage, + hugegraph_user=hg_user or None, + hugegraph_password=hg_password or None, + env=env, + ) + if env.get("HUGEGRAPH_URL"): + os.environ["HUGEGRAPH_URL"] = env["HUGEGRAPH_URL"] + elif storage != "hugegraph": + # only an explicit switch away from HugeGraph withdraws the endpoint + os.environ.pop("HUGEGRAPH_URL", None) + + async with self.sessionmaker() as session: + await save_runtime_config( + session, + llm_provider=provider, + llm_base_url=base_url or None, + llm_model=model or None, + llm_api_key=api_key or None, + storage_provider=storage, + # persist what the CONSOLE was told, not the resolved endpoint: + # a NULL here means "inherit HUGEGRAPH_URL from the deployment", + # which stays true even if the deployment later moves the graph + storage_url=storage_url or None, + storage_user=hg_user or None, + storage_password=hg_password or None, + ) + await session.commit() + + # Re-run provider registrations against the new settings, then rebuild + # the service objects that captured the old clients/endpoints. + old_llm = self.llm + if old_llm is not None and hasattr(old_llm, "aclose"): + try: + await old_llm.aclose() + except Exception: # noqa: BLE001 -- old client teardown is best-effort + pass + self.llm = None + self.graph_store_factory = None + if self.ext is not None: + self.ext.reload_providers() + if self.compiled is not None: + self._rebuild_services(self.compiled) + return await self.runtime_config() + + def _storage_provider(self) -> str: + """Effective storage provider, inferring HugeGraph from a configured URL. + + Tests and older deployments pass ``Settings(env={"HUGEGRAPH_URL": ...})`` + directly; those must keep working without an explicit storage-provider row. + """ + configured = self.settings.storage_provider or "sqlite" + if configured == "hugegraph": + return "hugegraph" + if self.settings.env.get("HUGEGRAPH_URL") or os.environ.get("HUGEGRAPH_URL"): + return "hugegraph" + return "sqlite" + + # ------------------------------------------------------- graph addressing + + def _graph_endpoint(self) -> str | None: + """The HugeGraph endpoint, whether or not a projection is declared. + + A declared projection names it as a DSL reference (``${HUGEGRAPH_URL}``); + without one the same variable is read straight from the resolution + scope. The console needs this to answer "does this domain already have a + graph?" *before* anything is declared — that answer is what decides + whether the graph is loaded or built. + """ + if self.compiled is not None and self.compiled.projections: + declared = next(iter(self.compiled.projections.values())) + try: + return self.settings.resolve_env_ref(declared.spec.endpoint) + except KeyError: + return None + return self.settings.env.get("HUGEGRAPH_URL") or os.environ.get("HUGEGRAPH_URL") or None + + def graph_name(self) -> str | None: + """This domain's one graph name (``None``: no package loaded yet).""" + if self.compiled is None: + return None + return domain_graph_name(self.compiled.package_name) + + async def graph_state(self) -> dict[str, Any]: + """Whether the domain's graph exists on the server: ``True``/``False``/``None``. + + The three-valued answer is the point. ``True`` means the graph is there + and its contents should be *loaded*; ``False`` means there is nothing to + load and building is the only way forward; ``None`` means we could not + find out (no endpoint, no graph extension, unreachable server) and the + caller must not pretend otherwise. + """ + name = self.graph_name() + if name is None: + return {"graph_name": None, "graphspace": None, "graph_exists": None} + graphspace = "DEFAULT" + if self.compiled is not None and self.compiled.projections: + graphspace = next(iter(self.compiled.projections.values())).spec.graphspace + if self._storage_provider() != "hugegraph" or self.graph_store_factory is None: + return {"graph_name": name, "graphspace": graphspace, "graph_exists": None} + endpoint = self._graph_endpoint() + if not endpoint: + return {"graph_name": name, "graphspace": graphspace, "graph_exists": None} + probe = self.graph_store_factory(endpoint, name, graphspace=graphspace, + http=self.projection_http, + user=self.settings.hugegraph_user, + password=self.settings.hugegraph_password, + data_dir=self.settings.hugegraph_data_dir) + try: + exists = await probe.exists() + except OOError as exc: + # a probe we were not allowed to make must not read as "absent": the + # console would offer a destructive build on a graph it cannot see + return {"graph_name": name, "graphspace": graphspace, "graph_exists": None, + "probe_error": exc.message} + return {"graph_name": name, "graphspace": graphspace, "graph_exists": exists, + "auth_rejected": bool(getattr(probe, "auth_rejected", False))} + + def _apply_package_root_override(self, compiled: CompiledOntology) -> CompiledOntology: + """Point a DB-loaded snapshot at the on-disk package when available.""" + candidate = self.package_root_override or self.package_root + if candidate and Path(candidate).is_dir() \ + and Path(compiled.package_root).resolve() != Path(candidate).resolve() \ + and _manifest_name(candidate) == compiled.package_name: + # only when the on-disk package IS the same package: a different + # domain activated from domains/ must keep its own root, or builder + # saves and promotions would land in the boot package's directory + compiled = replace(compiled, package_root=str(candidate)) + if not Path(compiled.package_root).is_dir(): + # a snapshot rebuilt from the registry has no directory; a domain + # activated from the domains root is deterministic to re-derive + dom = self.domains_root() / compiled.package_name + if dom.is_dir(): + compiled = replace(compiled, package_root=str(dom)) + return compiled + + async def publish(self, package_root: str, *, created_by: str | None = None) -> CompiledOntology: + self.package_root = package_root + pkg = load_package(package_root) + async with self.sessionmaker() as session: + compiled = await self.registry.publish(session, pkg, created_by=created_by or "ontogeny") + await session.commit() + self._rebuild_services(compiled) + async with self.sessionmaker() as session: + await self.repo.ensure(session) + return compiled + + async def reload(self) -> CompiledOntology: + """Refresh services from the registry (e.g. after an evolve promotion).""" + async with self.sessionmaker() as session: + compiled = await self.registry.load_latest(session) + if compiled is not None and ( + self.compiled is None or compiled.content_hash != self.compiled.content_hash + ): + self._rebuild_services(compiled) + async with self.sessionmaker() as session: + await self.repo.ensure(session) + return self.compiled # type: ignore[return-value] + + def _rebuild_services(self, compiled: CompiledOntology) -> None: + compiled = self._apply_package_root_override(compiled) + if not Path(compiled.package_root).is_dir() and compiled.functions: + log.warning( + "packages loaded from the registry have no directory on disk; " + "set ONTOGENY_PACKAGE_ROOT so function sources in %s can be resolved", + sorted(compiled.functions), + ) + self.compiled = compiled + # the MCP tool list is a compiled artifact: a promote that changes the + # ontology must be reflected at the next list_tools, not after a restart + surface = getattr(self, "mcp_surface", None) + if surface is not None: + surface.refresh() + self.repo = ObjectRepository(compiled) + self.sync_engine = SyncEngine( + compiled, self.settings, self.repo, + lambda store: make_source(store, self.settings.resolve_env_ref(store.spec.connection)), + ) + self.policy = PolicyEngine(compiled) + # sc.llm comes from an llm-provider extension (llm-ollama ships with the + # repo); None means "capability absent" and every LLM consumer degrades. + # The sandbox shares the platform LLM gateway: ontogeny.llm() inside a + # function is budgeted + routed here, never called from the child + self.sandbox = FunctionSandbox(compiled, self.repo, llm=self.llm, + llm_call_timeout_s=self.settings.llm_call_timeout_s) + # Both ways of having a body, behind one door: code runs in the sandbox, + # a declarative pipeline runs here through the same engine and gateway + # the sandbox reaches over RPC. Every caller uses `self.functions`, so + # the two cannot drift in budgeting, masking or error reporting. + self.functions = FunctionRuntime(self.sandbox, DeclarativeRuntime( + llm=self.llm, repo=self.repo, compiled=compiled, + http_allowlist=list(self.settings.function_http_allowlist or ()), + resolve_ref=self.settings.resolve_env_ref, + llm_call_timeout_s=self.settings.llm_call_timeout_s, + )) + self.derivation = DerivationWorker(compiled, self.repo, self.functions, self.sessionmaker) + self.runtime = ActionRuntime( + compiled, self.repo, self.policy, self.settings, + function_runner=self._function_runner, + ) + self.query = QueryService(compiled, self.repo, self.policy) + self.telemetry = TelemetryService(compiled) + self.eval_runner = EvalRunner(self.repo, self.query) + self.promoter = Promoter(self.settings, self.registry) + # evolution proposer: the LLM decides by default when a provider is + # configured (bounded by the mutation catalog + schema validation -- + # see ontogeny.evolve.proposer.DecidingProposer); without one the loop runs + # the deterministic heuristic proposer unchanged + from .evolve.proposer import DecidingProposer + self.proposer = DecidingProposer(self.llm) + + # optional projection layer (zero dependency unless declared AND configured; + # the graph store comes from a graph extension -- without one, or with an + # unresolvable endpoint, the platform degrades gracefully to SQL-only) + self.projection_worker = None + self.projection_client = None + self.projection_decl = None + # WHY the layer is not wired, as a stable code. "Declared" and "wired" + # are different questions, and a console that conflates them offers + # "create a projection" for a package that already has one -- an action + # the backend then refuses. Recording the blocking cause here is what + # lets the console show the fix instead of a dead end. + self.projection_blocked_by: str | None = None + provider = self._storage_provider() + if compiled.projections and provider != "hugegraph": + self.projection_blocked_by = "provider-sqlite" + log.info("projection %s declared but storage provider is %s; running SQL-only", + next(iter(compiled.projections.values())).metadata.name, provider) + if compiled.projections and provider == "hugegraph": + first = next(iter(compiled.projections.values())) + try: + endpoint = self.settings.resolve_env_ref(first.spec.endpoint) + except KeyError as exc: + self.projection_blocked_by = "endpoint-unresolved" + log.warning("projection %s declared but endpoint unresolved (%s); " + "running SQL-only, graph APIs disabled", first.metadata.name, exc) + else: + if self.graph_store_factory is None: + self.projection_blocked_by = "extension-missing" + log.warning("projection %s declared but no graph extension is loaded " + "(extensions/graph-hugegraph); running SQL-only", + first.metadata.name) + else: + self.projection_client = self.graph_store_factory( + endpoint, first.spec.graph, + graphspace=first.spec.graphspace, api=first.spec.api, + http=self.projection_http, + user=self.settings.hugegraph_user, + password=self.settings.hugegraph_password, + data_dir=self.settings.hugegraph_data_dir, + ) + self.projection_worker = ProjectionWorker(compiled, first, self.projection_client) + self.projection_decl = first + + + async def _projection_hook(event) -> bool: + if self.projection_worker is None: + return True + async with self.sessionmaker() as session: + return await self.projection_worker.handle_event(self.repo, session, event) + + async def _derivation_hook(event) -> bool: + return await self.derivation.handle_event(event) + + async def _derivation_drain() -> dict[str, int]: + return await self.derivation.drain() + + self.outbox = OutboxDispatcher( + self.sessionmaker, self.settings, sse=self.sse, projection_hook=_projection_hook, + derivation_hook=_derivation_hook, derivation_drain=_derivation_drain, + ) + self.agents = AgentBroker(self) + self.eval_runner.agent_broker = self.agents # agent eval cases use the same gates + # accounts & sessions: identity only -- authorization stays in the policy + # plane, which is why AuthService never decides anything about resources + self.auth = AuthService(self) + # agent engines come from extensions (agent-paradigm §3.1): sc.agent_engines + # maps engine kind -> factory(sc) -> engine, registered by the extension host + # at boot; dispatch in the /run endpoint stays untouched by new engines. + + async def _function_runner(self, session, entry: str, params: dict, ctx) -> list[dict]: + if self.compiled is None: + raise StoreError("service not initialized") + for fn in self.compiled.functions.values(): + if fn.spec.entry == entry: + return await self.sandbox.run(session, fn, params) + raise StoreError(f"function entry {entry!r} not found") + + # --------------------------------------------------------------- domains + + def domains_root(self) -> Path: + """Directory holding switchable domains. + + ``ONTOGENY_DOMAINS_ROOT`` wins; the default is a ``domains/`` sibling of the + *boot* package, which in the demo layout is ``<data-dir>/domains`` -- + next to ``pkg/`` (the boot copy) and ``pkg.pristine/``. Derived from the + boot package, never the current one: after activating a domain the + current package root lives INSIDE this directory. + """ + env = os.environ.get("ONTOGENY_DOMAINS_ROOT") + if env: + return Path(env) + base = Path(self.boot_package_root or self.package_root or ".").resolve().parent + if base.name == "domains": + # source-checkout layout: the boot package itself lives at + # <repo>/domains/<name>, so parent IS the domains root -- appending + # another "domains" produced <repo>/domains/domains/<name> + return base + return base / "domains" + + def list_domains(self) -> list[dict[str, Any]]: + """Every activatable domain: each domains-root subdirectory carrying an + ontology.yaml, plus the active package itself when it lives elsewhere + (the boot copy). The active domain sorts first.""" + active_name = self.compiled.package_name if self.compiled else None + candidates: list[Path] = [] + root = self.domains_root() + if root.is_dir(): + candidates.extend( + p for p in sorted(root.iterdir()) if (p / "ontology.yaml").is_file() + ) + for pkg_root in (self.boot_package_root, self.package_root): + path = Path(pkg_root) if pkg_root else None + if path is not None and (path / "ontology.yaml").is_file(): + candidates.append(path) + + out: list[dict[str, Any]] = [] + seen: set[str] = set() + for path in candidates: + manifest = _read_manifest(path) + meta = manifest.get("metadata") or {} + name = meta.get("name") or path.name + if name in seen: + continue + seen.add(name) + out.append({ + "name": name, + "display": meta.get("display"), + "description": meta.get("description"), + "path": str(path.resolve()), + "objects": _count_yaml(path / "objects"), + "links": _count_yaml(path / "links"), + "actions": _count_yaml(path / "actions"), + "check": _check_package(path), + "active": name == active_name, + }) + out.sort(key=lambda d: (not d["active"], d["name"])) + return out + + # ---- canvas layout sidecar ------------------------------------------ + # + # The ontology canvas' hand-arranged node positions used to live in the + # browser's localStorage; now they live WITH the package, as a `layout.yaml` + # sidecar next to ontology.yaml. Same trade as the resources themselves: + # per-domain by construction, it travels with zip import/export and Git, + # and it survives browsers and machines. The loader only scans the known + # resource directories, so the sidecar is invisible to the DSL. + + def storage_summary(self) -> dict[str, Any]: + """Where materialized object data physically lives, spelled out. + + The DSL declares a backing STORE (the external source of record, e.g. + the ERP/MES database); syncing PULLS rows from that source into the + platform's own object tables (``ontogeny_obj_<type>``) inside THIS + deployment's database (SQLite file in demos, Postgres in production). + Queries, actions, versions and audit all read those object tables — + never the source directly. This summary names the concrete target so + consoles can say exactly where the data landed.""" + dsn = self.settings.db_dsn or "" + if "sqlite" in dsn: + raw = dsn.split("///", 1)[-1] or "(in-memory)" + target = {"kind": "sqlite", "target": str(Path(raw).resolve()) if not raw.startswith(":") else raw} + elif dsn.startswith(("postgres", "postgresql")): + m = re.match(r"(?:postgresql?[+a-z]*://)([^:/@]+)(?::[^@]*)?@([^/]+)/([^?]+)", dsn) + shown = f"postgres://{m.group(1)}@{m.group(2)}/{m.group(3)}" if m else "(configured DSN)" + target = {"kind": "postgres", "target": shown} # password never leaves the process + else: + target = {"kind": "unknown", "target": "(unrecognized DSN)"} + + objects: dict[str, Any] = {} + if self.compiled is not None: + for name, obj in self.compiled.objects.items(): + backing = getattr(obj.spec, "backing", None) + sync = getattr(backing, "sync", None) if backing else None + store_res = self.compiled.stores.get(backing.store) if backing else None + desc: dict[str, Any] = { + "table": self.compiled.table_name(name), + "store": backing.store if backing else None, + "mode": backing.mode if backing else None, + "source_table": backing.source.table if backing and backing.source else None, + "store_type": store_res.spec.type if store_res else None, + } + if sync is not None: + desc["sync"] = sync.strategy + if sync.schedule: + desc["sync_schedule"] = sync.schedule + if sync.watermark is not None and sync.watermark.column: + desc["sync_watermark"] = sync.watermark.column + objects[name] = desc + + # declared sources with their RESOLVED, credential-free targets: the + # console can then say "sqlite · /path/erp.db" instead of a bare store + # name, which reads like a placeholder + stores: dict[str, dict[str, str]] = {} + for name, store in (self.compiled.stores.items() if self.compiled else []): + conn = store.spec.connection or "" + try: + conn = self.settings.resolve_env_ref(conn) + except Exception: # noqa: BLE001 -- an unresolved ${VAR} stays symbolic + pass + stores[name] = {"type": store.spec.type, "target": _mask_dsn(conn)} + + return {"target": target, "object_table_prefix": "ontogeny_obj_", "objects": objects, + "stores": stores} + + # canvas layout: a front-end concern, kept in its own module (ontogeny.layout) + def layout_path(self) -> Path | None: + return self._layout.path() + + def read_layout(self) -> dict[str, Any]: + return self._layout.read() + + def write_layout(self, layout: dict[str, Any]) -> None: + self._layout.write(layout) + + def create_domain(self, name: str, *, display: str | None = None, + description: str | None = None) -> dict[str, Any]: + """Scaffold a minimal, valid package under the domains root: manifest + + empty resource directories. Activation is the caller's decision.""" + if not _DOMAIN_NAME_RE.fullmatch(name): + raise DomainError(f"domain name must be kebab-case [a-z][a-z0-9-], got {name!r}") + if name.lower() in ("domains", "pkg", "pkg.pristine"): + raise DomainError(f"domain name {name!r} is reserved") + dom = self.domains_root() / name + if dom.exists(): + raise DomainError(f"domain {name!r} already exists", details={"path": str(dom)}) + for sub in _DOMAIN_DIRS: + (dom / sub).mkdir(parents=True, exist_ok=True) + import yaml + + manifest = { + "apiVersion": "ontogeny/v1", + "kind": "Ontology", + "metadata": { + "name": name, + "display": display or name, + "description": description, + "version": "0.1.0", + }, + "spec": {"imports": []}, + } + (dom / "ontology.yaml").write_text( + yaml.safe_dump(manifest, allow_unicode=True, sort_keys=False), encoding="utf-8", + ) + return { + "name": name, + "display": display or name, + "description": description, + "path": str(dom.resolve()), + "objects": 0, "links": 0, "actions": 0, + "active": False, + } + + async def import_domain(self, data: bytes, *, activate: bool = True) -> dict[str, Any]: + """Turn an uploaded domain-package zip into a switchable domain. + + The zip is treated as untrusted input end to end: entries must stay + inside the extraction directory (no absolute paths, no ``..``), the + archive is bounded (entry count + uncompressed size), and the package + must VALIDATE before anything is moved into the domains root -- a + rejected import leaves no residue. On success the folder lands at + ``<domains root>/<manifest name>`` and can be activated like any + scaffolded domain. + """ + import io + import shutil + import tempfile + import zipfile + + from .core.loader import load_package + from .core.validator import validate + + MAX_ENTRIES = 2000 + MAX_UNCOMPRESSED = 256 * 1024 * 1024 # 256 MB across all entries + + root = self.domains_root() + root.mkdir(parents=True, exist_ok=True) + + try: + zf = zipfile.ZipFile(io.BytesIO(data)) + except zipfile.BadZipFile as exc: + raise DomainError(f"not a zip archive: {exc}") from exc + + names = zf.namelist() + if not names: + raise DomainError("the archive is empty") + if len(names) > MAX_ENTRIES: + raise DomainError(f"too many entries (>{MAX_ENTRIES})") + + # every member must resolve INSIDE the extraction dir; a leading + # single folder is the normal "zip of the domain directory" shape and + # is unwrapped so the manifest sits at the extraction root + def _strip_root(ns: list[str]) -> tuple[str, int] | None: + first = ns[0].split("/")[0] + if first and all(n == first or n.startswith(first + "/") for n in ns): + return first + "/", len(first) + 1 + return None + + prefix, plen = _strip_root(names) or ("", 0) + rel_names = [n[plen:].rstrip("/") for n in names if n[plen:].rstrip("/")] + if "ontology.yaml" not in rel_names: + raise DomainError( + "no ontology.yaml found: the archive must contain a domain package " + "(a folder with ontology.yaml, or the package files at the zip root)" + ) + + staging = Path(tempfile.mkdtemp(prefix=".import-", dir=root)) + try: + total = 0 + for info in zf.infolist(): + member = info.filename[plen:] + if not member or member.endswith("/"): + continue + target = (staging / member).resolve() + if staging.resolve() not in target.parents: + raise DomainError(f"unsafe archive member: {info.filename!r}") + total += info.file_size + if total > MAX_UNCOMPRESSED: + raise DomainError("archive too large when extracted") + target.parent.mkdir(parents=True, exist_ok=True) + with zf.open(info) as src, target.open("wb") as dst: + shutil.copyfileobj(src, dst) + + # a package that does not validate can never become a domain: the + # switcher would list something activate_domain must then refuse + report = validate(load_package(str(staging))) + if not report.ok: + errs = "; ".join(f"{i.code}@{i.resource}: {i.message}" + for i in report.issues if i.severity == "error") + raise DomainError(f"package rejected by validator: {errs}", + details={"issues": [ + {"code": i.code, "resource": i.resource, + "message": i.message} for i in report.issues + ]}) + + name = (_read_manifest(staging).get("metadata") or {}).get("name") or "" + if not _DOMAIN_NAME_RE.fullmatch(name): + raise DomainError( + f"package name must be kebab-case [a-z][a-z0-9-], got {name!r}") + if name.lower() in ("domains", "pkg", "pkg.pristine"): + raise DomainError(f"domain name {name!r} is reserved") + final = root / name + if final.exists(): + raise DomainError(f"domain {name!r} already exists", + details={"path": str(final)}) + + staging.rename(final) + finally: + # staging was renamed on success; on any failure nothing remains + shutil.rmtree(staging, ignore_errors=True) + + if activate: + compiled = await self.activate_domain(final) + out = self._domain_entry(final, active=True) + out["content_hash"] = compiled.content_hash + return out + return self._domain_entry(final, active=False) + + def _domain_entry(self, path: Path, *, active: bool) -> dict[str, Any]: + """One list_domains-shaped record for a freshly imported domain.""" + meta = _read_manifest(path).get("metadata") or {} + return { + "name": meta.get("name") or path.name, + "display": meta.get("display"), + "description": meta.get("description"), + "path": str(path.resolve()), + "objects": _count_yaml(path / "objects"), + "links": _count_yaml(path / "links"), + "actions": _count_yaml(path / "actions"), + "active": active, + } + + def resolve_domain(self, name: str) -> Path: + """Directory of a named domain: the active or boot package when the + name matches one of them, else ``<domains root>/<name>``.""" + for pkg_root in (self.package_root, self.boot_package_root): + path = Path(pkg_root) if pkg_root else None + if path is not None and (path / "ontology.yaml").is_file() \ + and _manifest_name(path) == name: + return path + dom = self.domains_root() / name + if (dom / "ontology.yaml").is_file(): + return dom + raise NotFoundError(f"unknown domain: {name}") + + @staticmethod + def _validate_package_for_switch(package_root: str | Path, label: str) -> None: + """Fail a switch *before* anything is retired if a package is invalid. + + The target package is checked so a bad target can never deactivate the + current domain. The current package is checked too: switching away from + a package that can no longer be loaded would strand the operator in the + next domain with no way back. + """ + from .core.validator import validate as _validate + + pkg = load_package(str(package_root)) + report = _validate(pkg) + if report.ok: + return + errors = "; ".join( + f"{i.code}@{i.resource}: {i.message}" + for i in report.issues if i.severity == "error" + ) + raise DSLValidationError( + f"{label} package rejected by validator: {errors}", + details={"issues": [ + {"code": i.code, "resource": i.resource, "message": i.message} + for i in report.issues + ]}, + ) + + async def activate_domain(self, package_root: str | Path) -> CompiledOntology: + """Switch the active domain with a hard preflight on both sides. + + The requirement is asymmetric on purpose: a broken *target* must never + delete the current rows, and a broken *current* package must block the + switch rather than becoming a one-way door. + """ + target = Path(package_root) + current = Path(self.package_root) if self.package_root else None + if current is not None and current.is_dir() and current.resolve() != target.resolve(): + try: + self._validate_package_for_switch(current, "current domain") + except DSLValidationError: + raise + except Exception as exc: # noqa: BLE001 -- report as a switch guard + raise DSLValidationError(f"current domain cannot be validated: {exc}") from exc + self._validate_package_for_switch(target, "target domain") + pkg = load_package(str(target)) + async with self.sessionmaker() as session: + compiled = await self.registry.activate(session, pkg, created_by="console") + await session.commit() + self.package_root = str(target.resolve()) + self._rebuild_services(compiled) + async with self.sessionmaker() as session: + await self.repo.ensure(session) + return compiled + + # ------------------------------------------------------------ operations + + async def function_source_revisions(self, name: str, *, limit: int = 10) -> list[dict[str, Any]]: + """Audit rows for one function's code edits, newest first. + + Read from the same ``ontogeny_revision`` table the Audit page renders, so + "who changed this code, and when" has one answer rather than two. + """ + from sqlalchemy import select + + from .action.models import RevisionRow + + async with self.sessionmaker() as session: + rows = (await session.execute( + select(RevisionRow) + .where(RevisionRow.object_type == "Function", + RevisionRow.object_id == name, + RevisionRow.action == "edit-function-source") + .order_by(RevisionRow.id.desc()) + .limit(limit) + )).scalars().all() + return [ + {"id": r.id, "principal": r.principal, "at": r.created_at.isoformat(), + "version": (r.after or {}).get("version"), + "bytes": (r.after or {}).get("bytes"), + "message": r.message} + for r in rows + ] + + async def save_function_source( + self, name: str, source: str, *, principal: dict, + ) -> dict[str, Any]: + """Write a function's code, publish it, and record who did it. + + The console's code editor exists because the alternative — hand-editing a + file on the server — is worse: no syntax check before it lands, no audit + trail, no immediate feedback. What it does NOT change is the sandbox: the + code still runs in a child process with rlimits and the capabilities the + DSL declares. Editing the code is a privilege; the capability list is + still what bounds it. + + Saving never RUNS the code — trying it before committing is what + ``test_function_source`` is for, and keeping the two apart is what makes + a save's side effects (file, version, revision row) exactly predictable. + """ + import ast as _ast + import hashlib + + from .action.models import RevisionRow + from .errors import DSLValidationError, NotFoundError + from .functions.sandbox import function_source_path + + if self.compiled is None or self.package_root is None: + raise StoreError("no working package to write into") + fn = self.compiled.functions.get(name) + if fn is None: + raise NotFoundError(f"unknown function {name!r}") + if fn.spec.runtime != "python": + raise DSLValidationError( + f"function {name!r} is runtime={fn.spec.runtime}: its body IS the DSL, " + "edit the steps instead" + ) + # Same entry guard as the test path: a malformed entry is authoring + # input, which deserves a 400, not the SandboxError (a 500) that + # ``function_source_path`` would raise for it. + _file, _, func = (fn.spec.entry or "").partition(":") + if not func: + raise DSLValidationError( + f"entry must be 'file.py:function', got {fn.spec.entry!r}", + details={"entry": fn.spec.entry}, + ) + path = function_source_path(self.compiled, fn) + + # Syntax first: nothing is written until the text at least parses, so a + # half-typed save cannot leave the package un-importable. + try: + _ast.parse(source) + except SyntaxError as exc: + raise DSLValidationError( + f"python syntax error on line {exc.lineno}: {exc.msg}", + details={"line": exc.lineno, "offset": exc.offset, "text": (exc.text or "").rstrip()}, + ) from exc + + before = path.read_text(encoding="utf-8") if path.is_file() else None + if before == source: + # nothing changed: report success without a publish or a revision row, + # so the history stays a record of actual changes + return {"ok": True, "changed": False, "path": str(path), + "version": self.compiled.function_versions.get(name)} + + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(source, encoding="utf-8") + try: + compiled = await self.publish(str(self.package_root), created_by=f"console:{principal.get('id')}") + except Exception: + # the package no longer validates: put the old text back so the + # working copy stays publishable, and let the caller see why + if before is None: + path.unlink(missing_ok=True) + else: + path.write_text(before, encoding="utf-8") + raise + + version = compiled.function_versions.get(name) + async with self.sessionmaker() as session: + session.add(RevisionRow( + action="edit-function-source", + object_type="Function", + object_id=name, + principal=str(principal.get("id") or "?"), + params={"path": str(path)}, + before={"source": before, "bytes": len(before or "")}, + after={"source": source, "bytes": len(source), + "version": version, + "sha256": hashlib.sha256(source.encode()).hexdigest()[:12]}, + outcome="executed", + message=f"source edited in the console ({len(source)} bytes)", + )) + await session.commit() + + out: dict[str, Any] = { + "ok": True, "changed": True, "path": str(path), "version": version, + "content_hash": compiled.content_hash, "bytes": len(source), + } + return out + + async def test_function_source(self, name: str, source: str) -> dict[str, Any]: + """Run the editor's buffer ONCE, persisting nothing. + + The counterpart to ``save_function_source``: a syntax error here is a + returned failure rather than a refused save, a run that calls out really + calls out — but no file is written, nothing is published, and no + revision row exists afterwards. Inputs are derived the same way the old + opt-in smoke run derived them (``_smoke_params``), because a test with + invented-but-plausible inputs is the only kind worth reading. + """ + import ast as _ast + + from .errors import DSLValidationError, NotFoundError + + if self.compiled is None: + raise StoreError("no working package to test against") + fn = self.compiled.functions.get(name) + if fn is None: + raise NotFoundError(f"unknown function {name!r}") + if fn.spec.runtime != "python": + raise DSLValidationError( + f"function {name!r} is runtime={fn.spec.runtime}: its body IS the DSL, " + "edit the steps instead" + ) + # A malformed entry would only explode later, inside the sandbox spawn + # (``entry.split(":")``), surfacing as an opaque 500. Validate it here + # so the editor gets a readable 400 — mirroring exactly what + # ``function_source_path`` enforces on the save path, no stricter. + entry = fn.spec.entry or "" + _file, _, func = entry.partition(":") + if not func: + raise DSLValidationError( + f"entry must be 'file.py:function', got {entry!r}", + details={"entry": entry}, + ) + # Report a syntax error before spawning anything: the child would only + # fail to import the module, and the traceback says less than ast does. + try: + _ast.parse(source) + except SyntaxError as exc: + raise DSLValidationError( + f"python syntax error on line {exc.lineno}: {exc.msg}", + details={"line": exc.lineno, "offset": exc.offset, "text": (exc.text or "").rstrip()}, + ) from exc + + async with self.sessionmaker() as session: + params, sources = await self._smoke_params(session, fn) + try: + value = await self.functions.run_source(session, fn, source, params) + # the child answers over a JSON line, so the value is already + # wire-safe (dates arrived as strings, tensors do not exist here) + return {"ok": True, "value": value, + "params": params, "param_sources": sources} + except OOError as exc: + return {"ok": False, "code": exc.code, "message": exc.message, + "params": params, "param_sources": sources} + + async def _smoke_params(self, session, fn) -> tuple[dict[str, Any], dict[str, str]]: + """Parameter values for a smoke run, and where each came from. + + Invoking with ``{}`` was a design mistake: every function that HAS + required parameters — precisely the ones worth smoke-testing — failed + instantly with "missing 1 required positional argument", which says + nothing about the code. So values are derived instead, in the order that + makes the run most meaningful: + + 1. the declared ``default`` (the author already said what is sensible); + 2. **a real primary key** when the parameter names one (``equipment_id`` + → an actual row's id), so the function reads real data and its own + logic executes instead of bailing out on "not found"; + 3. a type-shaped sample (first enum value, 1, true, now, "sample"). + + The sources travel back to the editor: a smoke run is only trustworthy if + whoever reads it knows which inputs were invented. + """ + from .core.types import PropertyType + + params: dict[str, Any] = {} + sources: dict[str, str] = {} + # primary keys by property name, so a `*_id` parameter can be answered + # with a row that actually exists + pk_index: dict[str, str] = {} + for obj_name, obj in self.compiled.objects.items(): + pk_index.setdefault(obj.spec.primaryKey[0], obj_name) + + for name, pdef in fn.spec.parameters.items(): + if pdef.default is not None: + params[name] = pdef.default + sources[name] = "declared default" + continue + obj_type = pk_index.get(name) + if obj_type is not None: + rows, _total = await self.repo.query(session, obj_type, limit=1) + if rows: + params[name] = rows[0].get(name) + sources[name] = f"real id from {obj_type}" + continue + kind = PropertyType.parse(pdef.type) + if kind.kind == "enum" and kind.enum_values: + params[name], sources[name] = kind.enum_values[0], "first enum value" + elif kind.kind in ("integer", "decimal"): + params[name], sources[name] = 1, f"sample {kind.kind}" + elif kind.kind == "boolean": + params[name], sources[name] = True, "sample boolean" + elif kind.kind in ("date", "timestamp"): + import datetime as _dt + + params[name] = _dt.datetime.now(_dt.timezone.utc).isoformat() + sources[name] = "now" + else: + params[name], sources[name] = "sample", "sample text" + return params, sources + + async def sync(self, object_type: str) -> dict[str, Any]: + try: + async with self.sessionmaker() as session: + summary = await self.sync_engine.sync(session, object_type) + await session.commit() + return summary + except Exception as exc: + # The failed run's audit must outlive the rollback above: without + # this, ontogeny_sync_run only ever recorded successes and the failure + # half of the quarantine/fitness signal did not exist. + import datetime as _dt + + from .stores.sync import SyncRunRow + try: + strategy = self.sync_engine.strategy_for(object_type) + except Exception: # noqa: BLE001 -- label only, best effort + strategy = "snapshot" + async with self.sessionmaker() as s2: + s2.add(SyncRunRow( + object_type=object_type, strategy=strategy, status="error", + error=f"{type(exc).__name__}: {exc}", + finished_at=_dt.datetime.now(_dt.timezone.utc), + )) + await s2.commit() + raise + + async def evolve_promote(self, proposal_id: int) -> dict[str, Any]: + """The one promote implementation shared by REST and the CLI. + + Carries the full safety net: the post-merge hot-swap of the compiled + snapshot and the direction-F verification (a merge that was green + before and goes red after is rolled back without a human). The CLI + used to call promoter.promote() directly, promoting with none of it.""" + from sqlalchemy import select as _select + + if self.package_root is None: + raise DSLValidationError("promotion requires ONTOGENY_PACKAGE_ROOT (git working copy)") + async with self.sessionmaker() as s: + from .evolve.models import ProposalRow + + p = (await s.execute(_select(ProposalRow).where(ProposalRow.id == proposal_id))).scalar_one_or_none() + if p is None: + raise NotFoundError(f"proposal {proposal_id} not found") + pre_green = bool(p.eval_report and p.eval_report.get("passed")) + result = await self.promoter.promote(s, p, self.package_root, eval_report=p.eval_report) Review Comment: Fixed (commit 9cfda13): `Promoter.promote` now refuses to promote a proposal whose status is not `proposed` or `evaluated`. ########## server/ontogeny/stores/ddl.py: ########## @@ -0,0 +1,195 @@ +# Copyright 2026 Apache HugeGraph Authors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Dynamic DDL: object/link tables compiled from the ontology snapshot. + +Each ObjectType gets ``ontogeny_obj_<type>`` with typed property columns plus system +columns (``_rev/_valid_from/_valid_to/_synced_at``). Versioning is +append-only: updates close the current row and insert a successor; the +"alive current" state is exactly ``_valid_to IS NULL``. Derived properties are +never stored (computed at assembly time). +""" +from __future__ import annotations + +import datetime as _dt + +from sqlalchemy import ( + JSON, + BigInteger, + Boolean, + Column, + Date, + DateTime, + Float, + MetaData, + String, + Table, + Text, +) +from sqlalchemy.types import TypeDecorator + +from ..core.models import JoinTableJoin +from ..core.types import PropertyType + +SYSTEM_COLUMNS = ("_rev", "_valid_from", "_valid_to", "_synced_at") + + +class UTCDateTime(TypeDecorator): + """Timezone-aware UTC timestamps that survive a round trip. + + SQLite drops the offset, so a stored timestamp came back *naive* and every + timestamp comparison/derivation against `now()` (which is aware) blew up. + Normalizing on both sides keeps the platform's documented UTC convention + true regardless of the backing store. + """ + + impl = DateTime(timezone=True) + cache_ok = True + + def process_bind_param(self, value, dialect): + if value is None: + return None + if isinstance(value, _dt.date) and not isinstance(value, _dt.datetime): + value = _dt.datetime(value.year, value.month, value.day) + if value.tzinfo is None: + return value.replace(tzinfo=_dt.timezone.utc) + return value.astimezone(_dt.timezone.utc) + + def process_result_value(self, value, dialect): + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=_dt.timezone.utc) + return value.astimezone(_dt.timezone.utc) + +# One MetaData per ontology snapshot: two packages may both define an +# `equipment` table with different columns, and a republished package may change +# them. Sharing one global MetaData by table name silently returned the wrong +# Table object (Unconsumed column names on write). +_metadata_by_hash: dict[str, MetaData] = {} +_MAX_CACHED_SNAPSHOTS = 4 + + +def _metadata_for(compiled) -> MetaData: + key = getattr(compiled, "content_hash", "default") + md = _metadata_by_hash.get(key) + if md is None: + if len(_metadata_by_hash) >= _MAX_CACHED_SNAPSHOTS: + _metadata_by_hash.pop(next(iter(_metadata_by_hash))) + md = MetaData() + _metadata_by_hash[key] = md + return md + + +def sql_type(ptype: PropertyType): + k = ptype.kind + if k in ("string", "enum"): + return Text + if k == "integer": + return BigInteger + if k == "decimal": + return Float + if k == "boolean": + return Boolean + if k == "date": + return Date + if k == "timestamp": + return UTCDateTime() + if k in ("array", "struct", "vector", "geo-point", "geo-shape", "media", "document"): + return JSON + raise ValueError(f"no sql mapping for {k}") # pragma: no cover + + +def object_table(compiled, object_type: str) -> Table: + metadata = _metadata_for(compiled) + name = compiled.table_name(object_type) + cached = metadata.tables.get(name) + if cached is not None: + return cached + obj = compiled.objects[object_type] + pk_cols = {compiled.col(p) for p in obj.spec.primaryKey} + cols: list[Column] = [] + for prop, pdef in obj.spec.properties.items(): + # expression deriveds are computed at read time (no column); + # function deriveds are MATERIALIZED by the derivation worker (column) + if pdef.derived_kind() == "expr": + continue + cols.append(Column( + compiled.col(prop), sql_type(pdef.ptype()), + nullable=not pdef.required, + # composite PK: business key + _valid_from -> versions coexist + primary_key=compiled.col(prop) in pk_cols, + )) + cols += [ + Column("_rev", BigInteger, nullable=False), + Column("_valid_from", DateTime(timezone=True), nullable=False, primary_key=True), + Column("_valid_to", DateTime(timezone=True), nullable=True), + Column("_synced_at", DateTime(timezone=True), nullable=True), + ] + return Table(name, metadata, *cols) + + +def link_table(compiled, link_name: str) -> Table: + lnk = compiled.links[link_name] + if not isinstance(lnk.spec.join, JoinTableJoin): + raise ValueError(f"link {link_name} is not join-table backed") + metadata = _metadata_for(compiled) + name = compiled.link_table(link_name) + cached = metadata.tables.get(name) + if cached is not None: + return cached + return Table( + name, metadata, + # _valid_from is part of the key: without it, link_add -> link_remove -> + # link_add of the same pair collided with the soft-closed row's PK + # (objects solved this with (business key, _valid_from); links must too). + Column("src_id", String(500), primary_key=True), + Column("dst_id", String(500), primary_key=True), + Column("_valid_from", DateTime(timezone=True), nullable=False, primary_key=True), + Column("_valid_to", DateTime(timezone=True), nullable=True), + ) + + +def ensure_tables(conn, compiled) -> None: + """Create missing tables and add missing columns (additive migration). + + Additive evolution must just work: when the evolution loop auto-merges a new + optional property (T0), the object table has to gain that column, otherwise + the promoted schema would be unqueryable. Destructive changes (drops, type + changes) are deliberately NOT handled here -- they need `ontogeny migrate`. + """ + from sqlalchemy import inspect, text + + inspector = inspect(conn) + existing = set(inspector.get_table_names()) + tables: list[Table] = [ + object_table(compiled, t) for t, o in compiled.objects.items() if o.spec.backing is not None + ] + [ + link_table(compiled, lnk) for lnk in compiled.links + if isinstance(compiled.links[lnk].spec.join, JoinTableJoin) + ] + for t in tables: + if t.name not in existing: + t.create(conn) + continue + present = {c["name"] for c in inspector.get_columns(t.name)} + for column in t.columns: + if column.name in present or column.primary_key: + continue + ddl_type = column.type.compile(conn.dialect) + conn.execute(text(f'ALTER TABLE {t.name} ADD COLUMN {column.name} {ddl_type}')) Review Comment: Fixed (commit a39a838): column identifiers in `ALTER TABLE ... ADD COLUMN` are now quoted. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
