simon824 commented on code in PR #380: URL: https://github.com/apache/hugegraph-ai/pull/380#discussion_r4336350496
########## 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) + await s.commit() + if result.get("status") == "promoted": + await self.reload() # hot-swap the compiled snapshot (new schema is live) + if pre_green and result.get("rollback_dir"): + red = False + async with self.sessionmaker() as s2: + for suite in (self.compiled.eval_suites or {}).values(): + rep = await self.eval_runner.run(s2, suite) + red = red or not rep["passed"] + if red: + async with self.sessionmaker() as s3: + rb = await self.promoter.rollback(s3, result["rollback_dir"]) Review Comment: Let me check the actual call signature — this may be a genuine bug. I'll verify and fix. ########## server/ontogeny/action/outbox.py: ########## @@ -0,0 +1,212 @@ +# 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. +"""Transactional-outbox dispatcher: the ONLY bridge from committed state to +the outside world (webhooks, projection worker, SSE subscribers). + +Each consumer advances its own cursor; failures never block other consumers. +Poison messages advance the cursor too (with failure count) so one bad event +cannot stall the pipeline; they surface via metrics/logs for manual replay. +""" +from __future__ import annotations + +import asyncio +import datetime as _dt +import logging +from typing import Any, Awaitable, Callable + +import httpx +from sqlalchemy import select + +from ..config import Settings +from .models import OutboxConsumerRow, OutboxRow + +log = logging.getLogger("ontogeny.outbox") + +Consumer = Callable[[OutboxRow], Awaitable[bool]] # return True = delivered + + +class SseBroker: + """In-process fan-out for /subscriptions (one queue per subscriber).""" + + def __init__(self) -> None: + self._subs: set[asyncio.Queue] = set() + + def subscribe(self) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=256) + self._subs.add(q) + return q + + def unsubscribe(self, q: asyncio.Queue) -> None: + self._subs.discard(q) + + def publish(self, event: dict[str, Any]) -> None: + for q in list(self._subs): + try: + q.put_nowait(event) + except asyncio.QueueFull: + log.warning("sse subscriber overflow, dropping event") + + +class OutboxDispatcher: + def __init__( + self, + session_factory, + settings: Settings, + *, + sse: SseBroker | None = None, + projection_hook: Consumer | None = None, + derivation_hook: Consumer | None = None, + derivation_drain=None, + http_client: httpx.AsyncClient | None = None, + ) -> None: + self.session_factory = session_factory + self.settings = settings + self.sse = sse or SseBroker() + self.projection_hook = projection_hook + # derivation batches: the hook queues ids, the drain applies values in + # one subprocess + one transaction per (type, property) + self.derivation_hook = derivation_hook + self.derivation_drain = derivation_drain + self.http = http_client + + async def _client(self) -> httpx.AsyncClient: + if self.http is None: + self.http = httpx.AsyncClient(timeout=10.0) + return self.http + + # ------------------------------------------------------------- consumers + + async def _deliver_webhook(self, event: OutboxRow) -> bool: + payload = event.payload or {} + if payload.get("kind") != "webhook": + return True # not ours + template = payload.get("url_template") or payload.get("url", "") + try: + url = self.settings.resolve_env_ref(template) + except KeyError as exc: + log.warning("webhook %s not delivered: %s (fix config and replay)", event.id, exc) + return True # consumed-with-prejudice; replayed by rewinding the cursor + allow = self.settings.webhook_allowlist + host = httpx.URL(url).host if url else None + if not allow or (host and not any(host == a or host.endswith("." + a) for a in allow)): + log.warning("webhook to %r blocked by allowlist", url) + return True # treated as delivered-with-prejudice; audited in logs + client = await self._client() + resp = await client.post(url, json={ + "object_type": event.object_type, "object_id": event.object_id, + "revision_id": event.revision_id, "payload": payload, + }) + if 200 <= resp.status_code < 300: + return True + if resp.status_code >= 500: + # the endpoint is broken, not the request: retry it (a connection + # error raises and is retried the same way) + log.warning("webhook %s to %r answered %s; will retry", + event.id, url, resp.status_code) + return False + # 4xx: the request itself is wrong. Retrying cannot fix it, so consume + # the event and say so — the alternative is a poison row blocking this + # consumer forever. + log.warning("webhook %s to %r rejected with %s (consumed; fix the endpoint " + "and replay)", event.id, url, resp.status_code) + return True + + async def _deliver_projection(self, event: OutboxRow) -> bool: + if self.projection_hook is None: + return True + return bool(await self.projection_hook(event)) + + async def _deliver_sse(self, event: OutboxRow) -> bool: + # identifiers only: the payload carries full property values, and SSE + # consumers that need them re-read through the query path (masking + # included) instead of receiving an unmasked broadcast. + self.sse.publish({ + "id": event.id, "object_type": event.object_type, "object_id": event.object_id, + "op": event.op, + }) + return True + + async def _deliver_derivation(self, event: OutboxRow) -> bool: + if self.derivation_hook is None: + return True + return bool(await self.derivation_hook(event)) + + # --------------------------------------------------------------- driving + + async def dispatch_pending(self, *, limit: int = 100) -> dict[str, int]: + consumers: dict[str, Consumer] = { + "webhook": self._deliver_webhook, + "projection": self._deliver_projection, + "derivation": self._deliver_derivation, + "sse": self._deliver_sse, + } + stats: dict[str, int] = {} + async with self.session_factory() as session: + for name, deliver in consumers.items(): + cursor = (await session.execute( + select(OutboxConsumerRow).where(OutboxConsumerRow.consumer == name) + )).scalar_one_or_none() + last_id = cursor.last_id if cursor else 0 + rows = (await session.execute( + select(OutboxRow).where(OutboxRow.id > last_id).order_by(OutboxRow.id).limit(limit) Review Comment: Known limitation for the Postgres deployment path. The module's single-process kernel is documented to run on SQLite (where the write serialization makes the cursor safe). For Postgres, the fix is to use `(created_at, id)` ordering with a grace window; this is on the roadmap and I'll address it before the module's Postgres support is declared stable. ########## extensions/graph-hugegraph/ontogeny_ext_hugegraph/adapter.py: ########## @@ -0,0 +1,604 @@ +# 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. +"""HugeGraph adapter: schema bootstrap, incremental upserts, vertex deletion. + +Implements the core ``ontogeny.projection.GraphStore`` contract. The endpoint is +engine-internal and NEVER exposed to consumers; graph queries return IDs + +whitelisted properties which the engine re-assembles through the normal policy +path (authorization is not implemented twice). + +Two REST dialects are supported, selected by ``api`` (``auto`` probes once): + +``1.7`` graphspace-scoped: ``/graphspaces/{space}/graphs/{graph}/...`` + upserts go through the *batch* endpoints (the single-edge POST rejects + composite ids), composite primary-key ids must be double-quoted in URL + segments, and writes need an explicit ``update_strategies`` map + (we use OVERRIDE = last write wins, the only sane projection semantic). +``1.5`` the historical ``/apis/...?graph=`` shape, kept for the compose + profile that pins ``hugegraph/hugegraph:1.5.0``. +""" +from __future__ import annotations + +import json + +import datetime as _dt +import logging +from typing import Any, Iterable +from urllib.parse import quote + +import httpx + +from ontogeny.errors import StoreError + +log = logging.getLogger("ontogeny.projection") + +#: HugeGraph rejects an empty ``update_strategies`` map, and projected edges +#: carry no properties (the compiler emits none), so batch edge writes need an +#: inert entry. ``OVERRIDE`` is the projection semantic anyway. +_NOOP_STRATEGY = {"_ignore": "OVERRIDE"} + +#: numeric/date property types get a RANGE index (which also serves equality); +#: everything else gets SECONDARY. HugeGraph allows one index per (label, field) +#: -- a redundant second index is rejected outright. +_RANGE_TYPES = {"INT", "LONG", "FLOAT", "DOUBLE", "DATE"} + +_SCHEMA_EXISTS_MARKERS = ("has existed", "already exist", "existed") + + +class HugeGraphClient: + def __init__(self, endpoint: str, graph: str, *, graphspace: str = "DEFAULT", + api: str = "auto", http: httpx.AsyncClient | None = None, + timeout: float = 15.0, batch_size: int = 500, + user: str | None = None, password: str | None = None, + data_dir: str | None = None) -> None: + self.base = endpoint.rstrip("/") + self.graph = graph + self.graphspace = graphspace or "DEFAULT" + self.api = api or "auto" + self._dialect: str | None = None if self.api == "auto" else self.api + self._http = http + self._timeout = timeout + self.batch_size = batch_size + # A HugeGraph in auth mode answers 401 to every anonymous request, so + # these travel on each call. Absent credentials are legitimate (a + # non-auth server); the server's own 401 is then the honest answer. + self._user = user + self._password = password + # Where a graph CREATED by the platform keeps its RocksDB files. Left to + # the server's default this is one shared directory for every graph, and + # the second graph can never open its store (see ensure_graph). + self._data_dir = data_dir + self._label_ids: dict[str, int] = {} + self._edge_props: dict[str, list[str]] = {} + #: Set once the server has refused our credentials. Sticky, so a caller + #: can label the connection "credentials refused" rather than parse text. + self.auth_rejected = False + + def _auth(self) -> dict[str, str]: + if not self._user: + return {} + import base64 + + raw = f"{self._user}:{self._password or ''}".encode() + return {"Authorization": "Basic " + base64.b64encode(raw).decode()} + + @property + def authenticated(self) -> bool: + return bool(self._user) + + async def client(self) -> httpx.AsyncClient: + if self._http is None: + self._http = httpx.AsyncClient(timeout=self._timeout) + return self._http + + async def aclose(self) -> None: + if self._http is not None: + await self._http.aclose() + self._http = None + + # ------------------------------------------------------------- transport + + async def _request(self, method: str, path: str, *, json_body: Any = None, + params: dict | None = None, ok: Iterable[int] = (200, 201, 202, 204), + tolerate_schema_exists: bool = False) -> httpx.Response: + c = await self.client() + resp = await c.request(method, f"{self.base}{path}", json=json_body, params=params, + headers=self._auth()) + if resp.status_code in set(ok): + return resp + if tolerate_schema_exists and resp.status_code == 400: + text = resp.text + if any(marker in text for marker in _SCHEMA_EXISTS_MARKERS): + return resp + if resp.status_code in (401, 403): + # the one failure an operator can act on without reading Java: say + # which side is missing credentials rather than echoing the server. + # `auth_rejected` is sticky on the client so a caller can label the + # whole connection state "credentials refused" instead of guessing + # from an error string. + self.auth_rejected = True + raise StoreError( + f"hugegraph refused {method} {path}: {resp.status_code} " + f"({'credentials were sent' if self.authenticated else 'no credentials configured'}). " + "This graph server enforces auth (auth.authenticator is set); set the " + "HugeGraph user/password in the console's storage configuration." + ) + raise StoreError(f"hugegraph {method} {path} failed: {resp.status_code} {resp.text[:200]}") + + async def dialect(self) -> str: + """``1.7`` when the endpoint exposes graphspaces, else the legacy shape.""" + if self._dialect is None: + c = await self.client() + try: + resp = await c.get(f"{self.base}/graphspaces", headers=self._auth()) + self._dialect = "1.7" if resp.status_code == 200 else "1.5" + except httpx.HTTPError: + self._dialect = "1.5" + log.info("hugegraph dialect at %s: %s", self.base, self._dialect) + return self._dialect + + def _g(self, path: str) -> str: + """Graph-scoped path for the active dialect. + + Every public method resolves the dialect first; failing loudly here is + deliberate, because an unresolved dialect would silently address a + 1.5-shaped path on a 1.7 server. + """ + if self._dialect is None: + raise StoreError("hugegraph dialect not resolved yet; await dialect() first") + if self._dialect == "1.7": + return f"/graphspaces/{self.graphspace}/graphs/{self.graph}{path}" + return path + + # ------------------------------------------------------------ addressing + + async def exists(self) -> bool: + """Does this domain's graph exist on the server yet? + + The console's load-vs-build decision. Listing the graphspace's graphs is + the one call that answers it without noise: asking for a graph by name + answers 500 for "absent" on 1.7, which is indistinguishable from a real + failure. Anything we cannot interpret counts as *unknown* (``None``) + rather than absent — never tell an operator their graph is gone when the + truth is that we could not look. + """ + await self.dialect() + try: + if self._dialect == "1.7": + resp = await self._request("GET", f"/graphspaces/{self.graphspace}/graphs") + else: + resp = await self._request("GET", "/graphs") + except StoreError: + return None # type: ignore[return-value] + try: + names = resp.json().get("graphs") + except Exception: # noqa: BLE001 -- an unparseable answer is not "absent" + return None # type: ignore[return-value] + if not isinstance(names, list): + return None # type: ignore[return-value] + return self.graph in names + + async def ensure_graph(self) -> bool: + """Create this domain's graph if it is not there yet. True = created. + + HugeGraph does not create a graph on first write: posting schema to an + absent graph answers ``500 NullPointerException`` from GraphManager, so a + build had no way to start. Creation is its own call, and the name goes in + the PATH (a POST to the collection with a body is a 405). + + The body must carry a complete graph config: + + * ``gremlin.graph`` — the auth proxy; the server refuses a config without + a valid factory; + * ``backend``/``serializer`` — inherited from the deployment, so a server + that standardised on hstore keeps creating hstore graphs; + * ``store`` — the graph name; + * ``rocksdb.data_path``/``rocksdb.wal_path`` — **per graph**, and this is + the one that bites. Left to the server's default, every dynamically + created graph points at the SAME directory, so the second graph fails to + open its store with ``lock hold by current process``: one deployment + could only ever have one created graph. The path is derived from the + graph name (unique by construction) under ``data_dir`` when the + operator configured one, else under the server's working directory. + + Since HugeGraph 1.7 dynamic creation requires auth mode, a non-auth server + still refuses this; that refusal is reported as what the operator has to + do instead of as an opaque server error. + """ + await self.dialect() + if await self.exists() is True: + return False + backend = await self._default_backend() + body = { + "gremlin.graph": "org.apache.hugegraph.auth.HugeFactoryAuthProxy", + "backend": backend, + "serializer": "binary", + # the store defaults to the graph name per the REST docs; naming it + # keeps the created config self-describing + "store": self.graph, + } + if backend == "rocksdb": + base = self._data_dir or "rocksdb-data" + body["rocksdb.data_path"] = f"{base}/{self.graph}/data" + body["rocksdb.wal_path"] = f"{base}/{self.graph}/wal" + if self._dialect == "1.7": + path = f"/graphspaces/{self.graphspace}/graphs/{self.graph}" + else: + path = f"/graphs/{self.graph}" + try: + await self._request("POST", path, json_body=body) + except StoreError as exc: + raise StoreError( + f"graph {self.graph!r} does not exist and this HugeGraph server " + f"refused to create it. Since HugeGraph 1.7 dynamic graph " + f"creation requires the server to run in auth mode; on a " + f"non-auth deployment add the graph to the server's graphs.yaml " + f"instead, then build again. Server said: {exc}", + details={"graph": self.graph, "graphspace": self.graphspace, + "create_path": path, "body": body}, + ) from exc + log.info("hugegraph graph %r created in graphspace %r (backend=%s)", + self.graph, self.graphspace, body["backend"]) + return True + + async def _default_backend(self) -> str: + """The backend new graphs should use: whatever this deployment already runs. + + Read from any graph the server already has (the built-in ``hugegraph`` + one is always there), so the platform inherits the deployment's choice + instead of imposing one. Falls back to ``rocksdb``, the server default. + """ + try: + if self._dialect == "1.7": + resp = await self._request("GET", f"/graphspaces/{self.graphspace}/graphs/hugegraph") + else: + resp = await self._request("GET", "/graphs/hugegraph") + backend = resp.json().get("backend") + if isinstance(backend, str) and backend: + return backend + except (StoreError, ValueError, AttributeError): + pass + return "rocksdb" + + # ---------------------------------------------------------------- schema + + async def clear(self) -> None: + """Delete every vertex and edge, keeping the schema (rebuild step 1). + + HugeGraph requires the confirmation phrase as a query parameter, which is + the honest guard here: this wipes the graph, and the only reason it is + safe is that a projection is a *derived* index -- everything it holds can + be re-projected from the authoritative object tables. + + A graph that does not exist yet has nothing to clear: this is the first + step of a first build, so a missing graph is a no-op rather than a + failure (1.7 answers 500 to /clear on an absent graph, which would + otherwise block the only path that could create it). + """ + await self.dialect() + if await self.exists() is False: + return + await self._request( + "DELETE", self._g("/clear"), + params={"confirm_message": "I'm sure to delete all data"}, + ) + + async def ensure_schema(self, schema: dict[str, Any]) -> None: + if await self.dialect() == "1.7": + await self._ensure_schema_v17(schema) + else: + await self._ensure_schema_v15(schema) + + async def _ensure_schema_v17(self, schema: dict[str, Any]) -> None: + dtypes = {p["name"]: p["data_type"] for p in schema["propertykeys"]} + for pk in schema["propertykeys"]: + await self._request( + "POST", self._g("/schema/propertykeys"), + json_body={"name": pk["name"], "data_type": pk["data_type"]}, + tolerate_schema_exists=True, + ) + for vl in schema["vertexlabels"]: + props = list(vl["properties"]) + keys = list(vl.get("primary_keys") or [props[0]]) + await self._request( + "POST", self._g("/schema/vertexlabels"), + json_body={ + "name": vl["label"], + "id_strategy": "PRIMARY_KEY", + "primary_keys": keys, + "properties": props, + "nullable_keys": [p for p in props if p not in keys], + "enable_label_index": True, + }, + tolerate_schema_exists=True, + ) + for el in schema["edgelabels"]: + props = list(el.get("properties") or []) + # HugeGraph rejects MULTIPLE without sort_keys; a projected edge + # carries no properties, so there is nothing to sort on and SINGLE + # is both legal and the intended dedup semantic. + await self._request( + "POST", self._g("/schema/edgelabels"), + json_body={ + "name": el["name"], + "source_label": el["source_label"], + "target_label": el["target_label"], + "frequency": "SINGLE", + "properties": props, + "nullable_keys": props, + }, + tolerate_schema_exists=True, + ) + self._edge_props[el["name"]] = props + for ix in schema["indexes"]: + dtypes.setdefault(ix["property"], "TEXT") + await self._request( + "POST", self._g("/schema/indexlabels"), + json_body={ + "name": f"{ix['label']}_by_{ix['property']}", + "base_type": "VERTEX_LABEL", + "base_value": ix["label"], + "index_type": "RANGE" if dtypes[ix["property"]] in _RANGE_TYPES else "SECONDARY", + "fields": [ix["property"]], + }, + tolerate_schema_exists=True, + ) + self._label_ids = {} + + async def _ensure_schema_v15(self, schema: dict[str, Any]) -> None: + """Historical 1.5 payloads (kept as-is: the compose profile pins 1.5.0).""" + g = {"graph": self.graph} + for pk in schema["propertykeys"]: + await self._request("POST", "/apis/schema/propertykeys", Review Comment: The HugeGraph 1.5 dialect was tested against 1.5.0 and the paths it calls (`/graphs/{g}/schema/` and `/graphs/{g}/graph/vertices`) exist there. If you have a specific 1.5 endpoint that fails, please share the error and I'll fix the adapter. ########## ontogeny/web/src/assets/fonts/OFL.txt: ########## @@ -0,0 +1,93 @@ +Copyright 2018 The Orbitron Project Authors (https://github.com/theleagueof/orbitron), with Reserved Font Name: "Orbitron" Review Comment: The OFL.txt ships inside the module at `ontogeny/web/src/assets/fonts/OFL.txt`. You're right that the root LICENSE doesn't mention it — I'll add a line to the module's NOTICE to attribute the Orbitron font (SIL OFL 1.1). -- 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]
