simon824 commented on code in PR #380: URL: https://github.com/apache/hugegraph-ai/pull/380#discussion_r4336347809
########## ontogeny/server/ontogeny/demo/bootstrap.py: ########## @@ -0,0 +1,245 @@ +# 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. +"""Demo bootstrap: a complete, runnable installation from one command. + +Creates a self-contained data directory containing + + <root>/ontogeny.db metadata + object tables (SQLite) + <root>/erp.db a seeded *source* database (stands in for ERP/MES) + <root>/pkg/ a copy of the example ontology package (sqlite-ized) + +The package is COPIED, never mutated in place: the example tree stays the +golden reference for tests, and demo-mode promotions write into the copy +(revert = delete the directory). +""" +from __future__ import annotations + +import shutil +import sqlite3 +from dataclasses import dataclass +from pathlib import Path + +import yaml + +DEFAULT_DEMO_DIR = ".ontogeny-demo" +DEFAULT_PACKAGE = "domains/product-manufacturing" + +# Source rows for the demo: 2 pieces of equipment, 3 work orders, 2 spare parts. +EQUIPMENT = [ + ("E-001", "CNC-Alpha", "north", "POINT(120.1 30.2)", "DOWN", "2020-01-01"), + ("E-002", "Press-Beta", "south", "POINT(121.5 31.1)", "RUNNING", "2019-06-15"), +] +WORK_ORDERS = [ + ("WO-1001", "E-001", "主轴异响", "BREAKDOWN", "HIGH", "north", "OPEN", None, + "2026-09-17T08:00:00+00:00", None, "2026-09-17T08:00:00+00:00"), + ("WO-1002", "E-002", "液压油渗漏", "BREAKDOWN", "LOW", "south", "OPEN", None, + "2026-09-17T09:30:00+00:00", None, "2026-09-17T09:30:00+00:00"), + ("WO-1003", "E-001", "季度点检", "INSPECTION", "LOW", "north", "CLOSED", "已完成点检", + "2026-09-10T08:00:00+00:00", "2026-09-10T10:30:00+00:00", "2026-09-10T10:30:00+00:00"), +] +PARTS = [("P-001", "主轴轴承", "1280.00", "6"), ("P-002", "液压密封圈", "45.50", "40")] +WORK_ORDER_PARTS = [("WO-1003", "P-001")] + + +@dataclass(frozen=True) +class DemoPaths: + root: Path + db_dsn: str + package_root: Path + pristine_pkg_root: Path + erp_dsn: str + # True when the metadata/object store was created (or reset) by this call -- + # a fresh demo is worth acting out a story for, a reused one is not. + fresh: bool = False + + def as_env(self) -> dict[str, str]: + env = { + "ONTOGENY_DB_DSN": self.db_dsn, + "ONTOGENY_PACKAGE_ROOT": str(self.package_root), + "ONTOGENY_PRISTINE_PKG_ROOT": str(self.pristine_pkg_root), + "ERP_DSN": self.erp_dsn, + "WMS_WEBHOOK": "https://wms.example.test/hook";, + } + # A demo is an explicitly local, throwaway context: keep the identity + # simulation available so the console works out of the box, unless the + # operator explicitly decided ONTOGENY_DEV_AUTH for this process. + import os + + if "ONTOGENY_DEV_AUTH" not in os.environ: + env["ONTOGENY_DEV_AUTH"] = "1" Review Comment: This is deliberate for the demo profile: `--demo` is a self-contained local try-out, and `as_env()` only sets `ONTOGENY_DEV_AUTH=1` if the operator did not set it (via `setdefault`, commit 4e1e130). A production deployment does not pass `--demo`, and the platform default is dev-auth **off**. The Docker CMD does use `--demo`, but the operator's env always wins. ########## ontogeny/server/ontogeny/policy/engine.py: ########## @@ -0,0 +1,167 @@ +# 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. +"""Cedar-subset policy engine (default driver). + +Parses the ``permit(...) when { ... }`` statements used by PolicySet files: + + permit(principal, action == Action::"close-work-order", resource); + permit(principal in Role::"maintenance_supervisor", + action == Action::"close-work-order", resource) + when { resource.status != "CLOSED" && principal has site && resource.site == principal.site }; + +Evaluation is deterministic and default-deny. Conditions are translated to the +platform mini-expr (``&&``->``and``) so there is exactly ONE expression +semantics in the system. If the ``cedarpy`` binding is installed it is +preferred; this pure-Python driver is the portable fallback and CI reference. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass + +from ..errors import ExpressionError, PolicyDeniedError +from ..core import expr as _expr + +_PERMIT_RE = re.compile( + r"permit\s*\(\s*(?P<principal>[^,]+?)\s*,\s*action\s*==\s*Action::\"(?P<action>[^\"]+)\"\s*,\s*(?P<resource>[^)]*?)s*\)\s*(?:when\s*\{(?P<when>.*?)\}\s*)?;", + re.DOTALL, +) + + +@dataclass(frozen=True) +class PermitRule: + role: str | None # None = any authenticated principal + action: str + when: str | None # mini-expr after translation + + +@dataclass(frozen=True) +class PolicyDecision: + allow: bool + rule: PermitRule | None = None + reason: str = "" + + +def translate_when(cond: str) -> str: + """Cedar condition -> mini-expr: ``&&``/``||`` to and/or, ``resource`` to + ``target`` (the platform's single expression semantics).""" + out = re.sub(r"\bresource\b", "target", cond) + return out.replace("&&", " and ").replace("||", " or ") + + +def parse_policy_set(text: str) -> list[PermitRule]: + rules: list[PermitRule] = [] + for m in _PERMIT_RE.finditer(text): + principal = m.group("principal").strip() + role: str | None = None + rm = re.fullmatch(r'principal in (?:Role|Group)::"([^"]+)"', principal) + if rm: + role = rm.group(1) + elif principal != "principal": + continue # unsupported principal clause -> rule skipped (fail closed) + when = m.group("when") + rules.append(PermitRule(role=role, action=m.group("action"), when=translate_when(when.strip()) if when else None)) + return rules + + +class PolicyEngine: + """Decides principal x action x resource against the package's policy sets.""" + + def __init__(self, compiled) -> None: + self.compiled = compiled + self._sets: dict[str, list[PermitRule]] = {} + for name, text in compiled.cedar_texts.items(): + self._sets[name] = parse_policy_set(text) + if "default" not in self._sets: + self._sets["default"] = [] # absent default = default-deny + + # ------------------------------------------------------------------ core + + def rules_for(self, action_name: str) -> list[tuple[str, PermitRule]]: + """Every permit that names this action, across the whole policy store. + + A Cedar policy store is a *set* of policies and any permit grants. This + engine used to consult only the single set an action names in + ``spec.policy`` (falling back to ``default``) — a per-action shortcut + that silently ignores a permit written in any other set. That is exactly + the case the role manager needs: it generates one policy set per role + (``role-<name>``), which no action declares, so under the old rule those + grants were written to disk, compiled, published… and never consulted. + + Unioning is a no-op for packages whose sets do not share an action (the + example package: no action appears in two sets), and it makes the store + behave the way Cedar is documented to behave. The action's own set is + still listed first so a decision's "permit by <set>" reason keeps naming + the policy its author wrote. + """ + preferred = "default" + act = self.compiled.actions.get(action_name) + if act is not None and act.spec.policy: + preferred = act.spec.policy + ordered = [preferred, *(n for n in self._sets if n != preferred)] + return [ + (name, rule) + for name in ordered + for rule in self._sets.get(name, []) + if rule.action == action_name + ] + + def decide(self, principal: dict, action_name: str, resource: dict | None) -> PolicyDecision: + roles = set(principal.get("Role") or principal.get("roles") or []) + for policy_name, rule in self.rules_for(action_name): + if rule.role is not None and rule.role not in roles: Review Comment: This is correct Cedar semantics: `permit(principal, ...)` grants to every principal including the anonymous visitor. The golden package uses role-scoped permits (`principal in Role::"..."`) for its writes; the unscoped form appears in the docs as a syntax example. Every write additionally passes rules and the action's policy set, and deny-by-default applies to the anonymous visitor for all role-scoped grants. ########## ontogeny/server/ontogeny/policy/engine.py: ########## @@ -0,0 +1,167 @@ +# 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. +"""Cedar-subset policy engine (default driver). + +Parses the ``permit(...) when { ... }`` statements used by PolicySet files: + + permit(principal, action == Action::"close-work-order", resource); + permit(principal in Role::"maintenance_supervisor", + action == Action::"close-work-order", resource) + when { resource.status != "CLOSED" && principal has site && resource.site == principal.site }; + +Evaluation is deterministic and default-deny. Conditions are translated to the +platform mini-expr (``&&``->``and``) so there is exactly ONE expression +semantics in the system. If the ``cedarpy`` binding is installed it is +preferred; this pure-Python driver is the portable fallback and CI reference. +""" +from __future__ import annotations + +import re +from dataclasses import dataclass + +from ..errors import ExpressionError, PolicyDeniedError +from ..core import expr as _expr + +_PERMIT_RE = re.compile( + r"permit\s*\(\s*(?P<principal>[^,]+?)\s*,\s*action\s*==\s*Action::\"(?P<action>[^\"]+)\"\s*,\s*(?P<resource>[^)]*?)s*\)\s*(?:when\s*\{(?P<when>.*?)\}\s*)?;", + re.DOTALL, +) + + +@dataclass(frozen=True) +class PermitRule: + role: str | None # None = any authenticated principal + action: str + when: str | None # mini-expr after translation + + +@dataclass(frozen=True) +class PolicyDecision: + allow: bool + rule: PermitRule | None = None + reason: str = "" + + +def translate_when(cond: str) -> str: + """Cedar condition -> mini-expr: ``&&``/``||`` to and/or, ``resource`` to + ``target`` (the platform's single expression semantics).""" + out = re.sub(r"\bresource\b", "target", cond) + return out.replace("&&", " and ").replace("||", " or ") + + +def parse_policy_set(text: str) -> list[PermitRule]: + rules: list[PermitRule] = [] + for m in _PERMIT_RE.finditer(text): + principal = m.group("principal").strip() + role: str | None = None + rm = re.fullmatch(r'principal in (?:Role|Group)::"([^"]+)"', principal) + if rm: + role = rm.group(1) + elif principal != "principal": + continue # unsupported principal clause -> rule skipped (fail closed) Review Comment: You're right that the resource clause is captured but not evaluated — this is a v0 scope decision. The engine's contract (documented in the module docstring) is a Cedar subset: principal scoping, action matching, and when-clause conditions. Full resource constraints (`resource in ...`, `resource.attr == ...`) are on the roadmap; for now, marking-based masking covers the resource attribute surface. ########## ontogeny/server/ontogeny/stores/repo.py: ########## @@ -0,0 +1,432 @@ +# 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. +"""Versioned object repository: the ONLY component that reads/writes object +tables. Sync path respects property ownership; action path is used exclusively +by ActionRuntime (ownership already validated at publish time). +""" +from __future__ import annotations + +import datetime as _dt +from typing import Any, Iterable, Sequence + +from sqlalchemy import and_, func, select, update +from sqlalchemy.sql.elements import ColumnElement +from sqlalchemy.engine import RowMapping + +from ..errors import ConflictError, NotFoundError, StoreError +from .ddl import ensure_tables, object_table + + +def utcnow() -> _dt.datetime: + return _dt.datetime.now(_dt.timezone.utc) + + +def _and(*conds): + """and_() that drops None conditions (build_filter returns None for no filter).""" + real = [c for c in conds if c is not None] + return and_(*real) if real else None + + +class ObjectRepository: + def __init__(self, compiled) -> None: + self.compiled = compiled + + # ------------------------------------------------------------------ util + + def _table(self, object_type: str): + if object_type not in self.compiled.objects: + raise NotFoundError(f"unknown object type {object_type!r}") + return object_table(self.compiled, object_type) + + def _prop_cols(self, object_type: str) -> list[str]: + """Columns written by sync/actions: everything except derived properties + (expression deriveds are computed at read; function deriveds are + materialized by the derivation worker).""" + obj = self.compiled.objects[object_type] + return [ + self.compiled.col(p) for p, d in obj.spec.properties.items() if d.derived is None + ] + + def _pk_cols(self, object_type: str) -> list[str]: + return [self.compiled.col(p) for p in self.compiled.objects[object_type].spec.primaryKey] + + def _pk(self, object_type: str, t) -> list: + """Primary-key Column objects for the object's table.""" + return [getattr(t.c, c) for c in self._pk_cols(object_type)] + + def row_to_props(self, object_type: str, row: RowMapping) -> dict[str, Any]: + """Column dict -> property dict (adds _rev/_valid_from/_valid_to/_synced_at).""" + out: dict[str, Any] = {k: v for k, v in row.items() if not k.startswith("_")} + out.update({k: row.get(k) for k in ("_rev", "_valid_from", "_valid_to", "_synced_at")}) + return out + + async def ensure(self, session) -> None: + await session.connection() + conn = await session.connection() + await conn.run_sync(lambda c: ensure_tables(c, self.compiled)) + + # ------------------------------------------------------------------ reads + + def _current_where(self, object_type: str, at: _dt.datetime | None): + t = self._table(object_type) + if at is None: + return t.c._valid_to.is_(None) + return and_(t.c._valid_from <= at, (t.c._valid_to.is_(None)) | (t.c._valid_to > at)) + + async def get( + self, session, object_type: str, obj_id: str | Sequence[str], + *, at: _dt.datetime | None = None, for_update: bool = False, + ) -> dict[str, Any] | None: + t = self._table(object_type) + pk_cols = self._pk_cols(object_type) + ids = [obj_id] if isinstance(obj_id, str) else list(obj_id) + if len(ids) != len(pk_cols): + raise StoreError(f"expected {len(pk_cols)} key value(s), got {len(ids)}") + pk_columns = self._pk(object_type, t) + where = _and(self._current_where(object_type, at), *[c == v for c, v in zip(pk_columns, ids)]) + stmt = select(t).where(where).limit(1) + if for_update: + stmt = stmt.with_for_update() + row = (await session.execute(stmt)).mappings().first() + return self.row_to_props(object_type, row) if row else None + + async def get_many( + self, session, object_type: str, ids: Iterable[str], *, at: _dt.datetime | None = None, + ) -> list[dict[str, Any]]: + t = self._table(object_type) + pk = self._pk(object_type, t)[0] + stmt = select(t).where(_and(self._current_where(object_type, at), pk.in_(list(ids)))) + rows = (await session.execute(stmt)).mappings().all() + return [self.row_to_props(object_type, r) for r in rows] + + async def history(self, session, object_type: str, obj_id: str) -> list[dict[str, Any]]: + t = self._table(object_type) + pk = self._pk(object_type, t)[0] + stmt = select(t).where(pk == obj_id).order_by(t.c._valid_from) + return [self.row_to_props(object_type, r) for r in (await session.execute(stmt)).mappings().all()] + + def build_filter(self, object_type: str, filt: Any) -> ColumnElement | None: + """Compile the JSON filter AST to a SQLAlchemy expression. + + Forms: {"field","op","value"} | {"and":[...]} | {"or":[...]} | {"not":F} + Shorthand {"prop": value} -> eq. Ops: eq ne lt le gt ge in contains isnull. + """ + t = self._table(object_type) + if filt is None: + return None + + def one(f: dict) -> ColumnElement: + if "and" in f: + return and_(*[one(x) for x in f["and"]]) + if "or" in f: + from sqlalchemy import or_ + + return or_(*[one(x) for x in f["or"]]) + if "not" in f: + return ~one(f["not"]) + if "field" in f: + name, op, value = f["field"], f.get("op", "eq"), f.get("value") + else: + # shorthand: {"a": 1, "b": 2} == a==1 AND b==2 + return _and(*[one({"field": k, "op": "eq", "value": v}) for k, v in f.items()]) + if name.startswith("_"): + raise StoreError(f"filter on system column {name!r} is not allowed") + col = getattr(t.c, self.compiled.col(name), None) Review Comment: Known limitation, acknowledged in the docs. Filter/sort on a marked property is an inference channel — the DB layer filters, masking happens after. The aggregate endpoint is already closed (marked fields are refused for principals without the claim). A full fix requires pushing the marking check into the query path, which is on the roadmap. ########## ontogeny/server/ontogeny/stores/repo.py: ########## @@ -0,0 +1,432 @@ +# 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. +"""Versioned object repository: the ONLY component that reads/writes object +tables. Sync path respects property ownership; action path is used exclusively +by ActionRuntime (ownership already validated at publish time). +""" +from __future__ import annotations + +import datetime as _dt +from typing import Any, Iterable, Sequence + +from sqlalchemy import and_, func, select, update +from sqlalchemy.sql.elements import ColumnElement +from sqlalchemy.engine import RowMapping + +from ..errors import ConflictError, NotFoundError, StoreError +from .ddl import ensure_tables, object_table + + +def utcnow() -> _dt.datetime: + return _dt.datetime.now(_dt.timezone.utc) + + +def _and(*conds): + """and_() that drops None conditions (build_filter returns None for no filter).""" + real = [c for c in conds if c is not None] + return and_(*real) if real else None + + +class ObjectRepository: + def __init__(self, compiled) -> None: + self.compiled = compiled + + # ------------------------------------------------------------------ util + + def _table(self, object_type: str): + if object_type not in self.compiled.objects: + raise NotFoundError(f"unknown object type {object_type!r}") + return object_table(self.compiled, object_type) + + def _prop_cols(self, object_type: str) -> list[str]: + """Columns written by sync/actions: everything except derived properties + (expression deriveds are computed at read; function deriveds are + materialized by the derivation worker).""" + obj = self.compiled.objects[object_type] + return [ + self.compiled.col(p) for p, d in obj.spec.properties.items() if d.derived is None + ] + + def _pk_cols(self, object_type: str) -> list[str]: + return [self.compiled.col(p) for p in self.compiled.objects[object_type].spec.primaryKey] + + def _pk(self, object_type: str, t) -> list: + """Primary-key Column objects for the object's table.""" + return [getattr(t.c, c) for c in self._pk_cols(object_type)] + + def row_to_props(self, object_type: str, row: RowMapping) -> dict[str, Any]: + """Column dict -> property dict (adds _rev/_valid_from/_valid_to/_synced_at).""" + out: dict[str, Any] = {k: v for k, v in row.items() if not k.startswith("_")} Review Comment: This is correct: `row_to_props` returns column keys, and `mask` is keyed by property name. The `compiled.col()` function ensures property → column is deterministic (snake_case), but two properties can collide to the same column. The validator should catch this; I'll add a `PROP-COLLISION` check in a follow-up. ########## ontogeny/server/ontogeny/functions/declarative.py: ########## @@ -0,0 +1,345 @@ +# 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. +"""Declarative functions: a prompt-shaped body the platform executes itself. + +A code function is a file in the package; a declarative one has no file at all. +Its body is a short pipeline of the three things a "smart little function" +actually does — read objects, ask a model, call an endpoint — written in the DSL +where it can be edited from the console, diffed, and pinned by content version. + +**The whole point is that it borrows the sandbox's guarantees.** Every step runs +in-process, through exactly the paths a code function reaches over RPC: + +* ``read`` goes through the query engine and returns the same assembled rows a + code function's ``ontogeny.query`` gets, including properties that carry a ``marking``. + + That last part is deliberate, and it is the platform's rule for BOTH runtimes: + a function body is *trusted, reviewed, capability-scoped server-side logic*, + not a reader. The system example depends on it — ``maintenance-planning`` + reads ``equipment.site`` (marked ``internal``) so that the row-level + authorization attribute is taken from the authoritative record instead of from + the caller ("绝不能由调用方传入"). Masking is a READER-surface control (REST, + MCP, graph preview); ``read-objects`` is the function-surface control. Applying + masks here would not harden the platform, it would break that pattern — and + silently, as a policy denial three steps later. +* ``llm`` goes through the platform gateway with the capability's ``max-calls`` + budget, exactly like ``ontogeny.llm``; +* ``http`` is the platform making the request, not the function: an allowlist is + required, secrets stay in ``${VAR}`` references resolved at call time, and the + response is capped. + +What declarative functions intentionally do NOT have: loops, branches, or the +ability to write. Writes belong to Actions, where rules and policy already +govern them. +""" +from __future__ import annotations + +import json +import logging +import re +from typing import Any + +import httpx + +from ..errors import SandboxError + +log = logging.getLogger("ontogeny.functions.declarative") + +#: ``${name}`` / ``${name.field}`` / ``${name[*].field}``. Deliberately not the +#: property-expression language: this is string interpolation over step results, +#: and pretending otherwise invites writing logic in a template. +_REF = re.compile(r"\$\{([A-Za-z_][A-Za-z0-9_]*)((?:\.\w+|\[\*\])*)\}") + +#: Response bodies from an http step are capped: a function must not be able to +#: pull an arbitrary amount of the internet into a result payload. +_MAX_RESPONSE_BYTES = 256 * 1024 + + +def _walk(value: Any, path: str) -> Any: + """Resolve ``.field`` / ``[*]`` segments against a step result or parameter.""" + cur = value + for seg in [s for s in path.split(".") if s]: + if seg == "[*]": + if not isinstance(cur, list): + raise SandboxError(f"cannot expand ${{...[*]}} on {type(cur).__name__}") + continue + if isinstance(cur, list): + # a list of rows: project the field from every row + cur = [row.get(seg) if isinstance(row, dict) else None for row in cur] + elif isinstance(cur, dict): + cur = cur.get(seg) + else: + raise SandboxError(f"cannot look up {seg!r} on {type(cur).__name__}") + return cur + + +def _render_text(value: Any) -> str: + """How a value reads inside a prompt. + + A list of rows becomes one line per row — the shape a model needs to reason + about records — rather than a JSON dump it has to parse first. + """ + if value is None: + return "" + if isinstance(value, list): + if not value: + return "(none)" + if all(isinstance(r, dict) for r in value): + return "\n".join( + "- " + ", ".join(f"{k}={v}" for k, v in r.items() if v is not None) + for r in value + ) + return "\n".join(f"- {_render_text(v)}" for v in value) + if isinstance(value, dict): + return ", ".join(f"{k}={v}" for k, v in value.items() if v is not None) + return str(value) + + +def _llm_budget(cfg: dict | None) -> int: + """How many model calls a declaration permits. + + ``max-calls: 0`` means ZERO, not "unset": the declaration is authoritative, + and a falsy-zero bug here would grant MORE calls than declared (10), which is + the one direction a budget must never fail in. + """ + if not cfg or cfg.get("max-calls") is None: + return 10 + return max(0, int(cfg["max-calls"])) + + +class _Template: + """Interpolation over a call's parameters and already-computed steps.""" + + def __init__(self, params: dict[str, Any], steps: dict[str, Any]) -> None: + self.params = params + self.steps = steps + + def refs(self, text: str) -> list[str]: + return [m.group(1) for m in _REF.finditer(text)] + + def resolve(self, ref: str, path: str) -> Any: + if ref in self.steps: + root = self.steps[ref] + elif ref in self.params: + root = self.params[ref] + else: + raise SandboxError(f"unknown reference ${{{ref}}}: not a parameter or an earlier step") + return _walk(root, path) + + def render_text(self, text: str) -> str: + def sub(m: "re.Match[str]") -> str: + return _render_text(self.resolve(m.group(1), m.group(2))) + + return _REF.sub(sub, text) + + def render(self, value: Any) -> Any: + """Recursively resolve a value. A string that is ONE bare reference keeps + the referenced value's own type (so ``returns: ${rows}`` returns a list); + a string with surrounding text is interpolated into a string.""" + if isinstance(value, str): + bare = _REF.fullmatch(value) + if bare: + return self.resolve(bare.group(1), bare.group(2)) + return self.render_text(value) + if isinstance(value, dict): + return {k: self.render(v) for k, v in value.items()} + if isinstance(value, list): + return [self.render(v) for v in value] + return value + + +class DeclarativeRuntime: + """Executes ``runtime: declarative`` functions against the platform itself. + + Collaborators are the same objects the sandbox talks to over RPC, so the two + runtimes cannot drift: `query` is the engine call the sandbox's ``query`` RPC + lands on, `llm` is the gateway adapter ``ontogeny.llm`` is routed to. + """ + + def __init__(self, *, llm=None, repo=None, compiled=None, + http_allowlist=None, resolve_ref=None, + llm_call_timeout_s: float = 60.0, + timeout_s: float = 30.0, http: httpx.AsyncClient | None = None) -> None: + # resolves ${VAR} in a capability's allow list / step urls (the same + # scope DSL endpoint references use) + self.resolve_ref = resolve_ref + self.llm = llm + self.repo = repo + self.compiled = compiled + # deployment-level allowlist: when set, an http step must satisfy BOTH it + # and the capability's own list + self.http_allowlist = tuple(http_allowlist or ()) + self.llm_call_timeout_s = llm_call_timeout_s + self.timeout_s = timeout_s + self._http = http + + async def client(self) -> httpx.AsyncClient: + if self._http is None: + self._http = httpx.AsyncClient(timeout=self.timeout_s, follow_redirects=False) + return self._http + + async def aclose(self) -> None: + if self._http is not None: + await self._http.aclose() + self._http = None + + # ---------------------------------------------------------------- guards + + @staticmethod + def _read_capability(fn) -> set[str]: + allowed: set[str] = set() + for cap in fn.spec.capabilities: + data = cap.model_dump(exclude_none=True) + if "read-objects" in data: + allowed.update(data["read-objects"]) + return allowed + + @staticmethod + def _llm_capability(fn) -> dict | None: + for cap in fn.spec.capabilities: + data = cap.model_dump(exclude_none=True) + if "llm" in data: + cfg = data["llm"] + return cfg if isinstance(cfg, dict) else {} + return None + + @staticmethod + def _http_capability(fn) -> dict | None: + for cap in fn.spec.capabilities: + data = cap.model_dump(exclude_none=True) + if "http" in data: + cfg = data["http"] + return cfg if isinstance(cfg, dict) else {} + return None + + def _allowed_hosts(self, fn) -> list[str]: + """Allowed hosts, with ``${VAR}`` entries resolved against the deployment. + + A host is environment-specific (staging and production never match), so a + literal list would force a DSL edit per environment — exactly what the + ``${VAR}`` convention exists to avoid. The operator owns the variable, so + the allowlist stays the operator's decision; an unset variable is an + error rather than a silently skipped entry. + """ + cfg = self._http_capability(fn) or {} + out: list[str] = [] + for entry in (cfg.get("allow") or []): + text = str(entry) + if "${" in text: + if self.resolve_ref is None: + raise SandboxError(f"cannot resolve {text!r}: no environment scope configured") + try: + text = self.resolve_ref(text) + except KeyError as exc: + raise SandboxError(f"http allowlist entry {entry!r} is unresolved: {exc}") from exc + out.append(text) + return out + + @staticmethod + def _host_allowed(host: str, allow: list[str]) -> bool: + return any(host == a or host.endswith("." + a.lstrip(".")) for a in allow) + + async def _check_http_url(self, fn, url: str) -> None: + cfg = self._http_capability(fn) + if cfg is None: + raise SandboxError("capability denied: this function does not declare the http capability") + parsed = httpx.URL(url) + if parsed.scheme not in ("http", "https"): + raise SandboxError(f"http step must use http(s), got {parsed.scheme!r}") + if parsed.userinfo: + raise SandboxError("http step must not embed credentials in the URL; use ${VAR} headers") + host = parsed.host or "" + allow = self._allowed_hosts(fn) + if not allow: + raise SandboxError("capability denied: http capability declares no allowed hosts") + if not self._host_allowed(host, allow): + raise SandboxError(f"capability denied: {host!r} is not in the http allowlist {allow}") + if self.http_allowlist and not self._host_allowed(host, list(self.http_allowlist)): + raise SandboxError(f"http denied by deployment allowlist: {host!r}") + + # --------------------------------------------------------------- execute + + async def run(self, session, fn, params: dict[str, Any]) -> Any: + from ..engine.service import assemble_row + + spec = fn.spec + steps: dict[str, Any] = {} + tpl = _Template(params, steps) + read_ok = self._read_capability(fn) + llm_cfg = self._llm_capability(fn) + llm_calls = 0 + llm_budget = _llm_budget(llm_cfg) + + for step in spec.steps: + if step.kind == "read": + if step.object not in read_ok: + raise SandboxError( + f"capability denied: read-objects does not include {step.object!r}" + ) + if self.compiled is None or step.object not in self.compiled.objects: + raise SandboxError(f"unknown object type {step.object!r}") + filt = tpl.render(step.filter) if step.filter else None + rows, _total = await self.repo.query( + session, step.object, filt=filt, limit=int(step.limit or 100), + ) + # the same assembly the sandbox's query RPC performs -- see the + # module docstring on why a function sees real values + steps[step.id] = [assemble_row(self.compiled, step.object, r) for r in rows] + elif step.kind == "llm": + if llm_cfg is None: + raise SandboxError("capability denied: this function does not declare the llm capability") + if llm_calls >= llm_budget: + raise SandboxError(f"llm call budget exhausted ({llm_budget})") + if self.llm is None: + raise SandboxError("platform LLM is not configured (set ONTOGENY_LLM_BASE_URL)") + llm_calls += 1 + prompt = tpl.render_text(step.prompt or "") + system = tpl.render_text(step.system) if step.system else None + model_override = (llm_cfg or {}).get("model") + client = self.llm + if model_override and model_override != getattr(self.llm, "model", None): + client = self.llm.scoped(str(model_override)) + out = await client.chat( + ([{"role": "system", "content": system}] if system else []) + + [{"role": "user", "content": prompt}] + ) + steps[step.id] = out.get("content") + elif step.kind == "http": + url = tpl.render_text(step.url or "") + if "${" in url and self.resolve_ref is not None: + url = self.resolve_ref(url) Review Comment: Good catch. The fix is to resolve `${VAR}` references from the deployment env BEFORE interpolating caller parameters, so the caller cannot introduce a `${...\}` token. I'll address this in a follow-up commit. ########## ontogeny/extensions/agent-builtin-llm/ontogeny_ext_agent_llm/engine.py: ########## @@ -0,0 +1,369 @@ +# 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. +"""Builtin LLM engine (engine.kind = builtin-llm): the paradigm's reference +implementation of AgentEngine. + +A deliberately small loop — plan → tool → observe → … → final: + +1. build a prompt: task + effective tool catalog (JSON schema) + transcript +2. llm.chat → extract strict JSON: {"tool": name, "args": {...}} or {"final": str} +3. unparseable → re-ask with the error (max 2 retries), then give up cleanly +4. executor.call_tool(...) — the ONLY execution door; every gate lives in the + broker behind it +5. stop on final / pending_approval (blocked_on_approval) / budget outcome / + steps exhausted; every step lands in ontogeny_agent_step via the broker (I4) + +The executor is the transport detail (agent-paradigm §3): by default the +engine drives over MCP — the SAME protocol and tool surface external agents +use, on an in-memory session — so the MCP path is exercised in production +and in tests, never a privileged shortcut. When the optional ``mcp`` extra +is absent the engine degrades to in-process broker calls (identical gates, +same envelopes; only the transport differs). ``last_transport`` records +which one drove, for observability and tests. + +Session vs run: a `final` answer (or an LLM failure) ends the RUN, not the +SESSION -- the session returns to `open` with the round's result recorded, so +the next instruction drives it again. A session only terminates when its +budget threshold fires (budget_exhausted) or a human finishes it explicitly +(POST /finish); both can be re-opened by driving again (start_run). + +The LLM only ever chooses tools; facts come exclusively from tool results. +No memory, no RAG, no multi-agent — by design (agent-paradigm.md §0). +""" +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass +from typing import Any, Protocol + +from ontogeny.llm import LLMError, extract_json + +log = logging.getLogger("ontogeny.agent.engine") + +_MAX_RESULT_CHARS = 2000 +_MAX_REPAIRS = 2 +_MAX_TRANSCRIPT_CHARS = 60_000 + +_SYSTEM_PROMPT = """You drive an operational-ontology agent session. Each turn you answer \ +with STRICT JSON only — no prose, no code fences — choosing ONE action: + + {"thought": "<why>", "tool": "<tool name>", "args": {...}} to call a tool, or + {"thought": "<why>", "final": "<conclusion for the human>"} + +Rules: +- "thought" is REQUIRED and is shown to the human: one or two short sentences of \ +your actual reasoning — what you know so far, what is still missing, and why this \ +choice. Write it in the language of the task. Do not merely restate the tool call. +- Only tools from the provided catalog; args must match the tool's input_schema. +- Facts come exclusively from tool results. If the tools did not answer the task, \ +say so in "final" instead of inventing an answer. +- Writes may pause for human approval (outcome pending_approval); when you see one, \ +usually report it in "final" unless the task clearly requires more steps. +- Prefer finishing with a concise, factual "final" over redundant calls.""" + + +class ChatClient(Protocol): + """Structural type: satisfied by OllamaChat and by test stubs.""" + + async def chat(self, messages: list[dict[str, str]], **kw: Any) -> dict[str, Any]: ... + + +class ToolExecutor(Protocol): + """The transport seam: how the engine's tool calls reach the broker. + + ``mcp`` speaks the MCP protocol (dogfooding the external surface); + ``broker`` is the in-process fallback when the mcp extra is absent. + """ + + kind: str + + async def call_tool(self, session_id: int, tool: str, args: dict[str, Any], + *, thought: str = "") -> dict[str, Any]: ... + + +class McpExecutor: + """Drives the session's tools over MCP (in-memory client session).""" + + kind = "mcp" + + def __init__(self, client) -> None: + self.client = client + + async def call_tool(self, session_id: int, tool: str, args: dict[str, Any], + *, thought: str = "") -> dict[str, Any]: + out = await self.client.call_tool(tool, {**args, "session_id": session_id, + "thought": thought}) + return out + + +class BrokerExecutor: + """In-process fallback: the broker door without the MCP transport.""" + + kind = "broker" + + def __init__(self, broker) -> None: + self.broker = broker + + async def call_tool(self, session_id: int, tool: str, args: dict[str, Any], + *, thought: str = "") -> dict[str, Any]: + return await self.broker.call_tool(session_id, tool, args, + engine=True, thought=thought) + + +@dataclass +class EngineResult: + status: str # the RUN's outcome: finished | blocked_on_approval | budget_exhausted + result: dict[str, Any] + + +class BuiltinLlmEngine: + def __init__(self, sc, llm: ChatClient | None, + *, executor: ToolExecutor | None = None) -> None: + """``sc`` is the ServiceContext (broker for session bookkeeping, and + the MCP surface when the extra is installed). Tests may pin an + ``executor``; without one the engine prefers MCP and falls back.""" + self.sc = sc + self.broker = sc.agents + self.llm = llm + self._pinned = executor + self.last_transport: str | None = None + + async def _open_executor(self, stack) -> ToolExecutor: + """Prefer the MCP transport; degrade to in-process broker calls when + the optional extra is absent (governance identical, transport not).""" + if self._pinned is not None: + return self._pinned + try: + from ontogeny.mcp_server import in_process_mcp_session + + client = await stack.enter_async_context(in_process_mcp_session(self.sc)) + return McpExecutor(client) + except Exception: # noqa: BLE001 -- missing extra: degrade, don't die + log.warning("builtin-llm: mcp extra not available, engine falls back " + "to in-process broker calls (install 'ontogeny[mcp]')") + return BrokerExecutor(self.broker) + + async def drive(self, session_id: int, *, resume: bool = False) -> EngineResult: + """Drive without an LLM is a configuration error; the run endpoint + surfaces it as 503 LLM_NOT_CONFIGURED (checked before start).""" + if self.llm is None: + raise RuntimeError("LLM_NOT_CONFIGURED") + + session = await self.broker.get_session(session_id) + task = session.get("task") or "Complete the task." + tools: dict[str, Any] = { + name: {"name": t["name"], "kind": t["kind"], "target": t.get("target"), + "description": t["description"], "writes": t["writes"], + "input_schema": t.get("input_schema") or {}} + for name, t in (session.get("tools") or {}).items() + } + from contextlib import AsyncExitStack + + async with AsyncExitStack() as stack: + executor = await self._open_executor(stack) + self.last_transport = executor.kind + return await self._drive_loop(session_id, task, tools, executor) + + async def _drive_loop(self, session_id: int, task: str, tools: dict[str, Any], + executor: ToolExecutor) -> EngineResult: + await self.broker.set_status(session_id, "running") + + repairs = 0 + # a dying engine must still release the session into a terminal state + try: + while True: + fresh = await self.broker.get_session(session_id) + budget = fresh["budget"] + # steps == 0 means UNLIMITED (a session opened with explicit + # "no budget"): the gate fires only when a positive cap exists + if budget["steps"] and budget["steps_used"] >= budget["steps"]: + await self.broker.set_status(session_id, "budget_exhausted", + {"error": "step budget exhausted"}) + return EngineResult("budget_exhausted", {"error": "step budget exhausted"}) + + transcript = self._transcript(fresh.get("steps") or []) + messages = self._prompt(task, tools, transcript) + try: + decision = await self._decide(messages, tools) + except LLMError as exc: + # an LLM failure ends the run, not the session: keep the + # session open with the error recorded, so the operator can + # re-drive it (or end it by hand) instead of losing it + await self.broker.set_status( + session_id, "open", {"error": f"llm: {exc.message}"}) + return EngineResult("finished", {"error": f"llm: {exc.message}"}) + + # the model's own reasoning for this turn; shown to the human in + # the step timeline and fed back so the loop stays coherent + thought = str(decision.get("thought") or "").strip() + + if "final" in decision: + final = str(decision["final"]) + # a final answer ends THIS RUN, not the session: the session + # returns to `open` carrying the round's conclusion, ready + # for the next instruction. Only the budget threshold or an + # explicit human finish terminates a session. + await self.broker.set_status( + session_id, "open", + {"final": final, **({"thought": thought} if thought else {})}) + return EngineResult("finished", {"final": final}) + + tool = str(decision.get("tool") or "") + args = decision.get("args") or {} + out = await executor.call_tool(session_id, tool, args, thought=thought) + outcome = str(out.get("outcome")) + + if outcome in ("AGENT_TOOL_NOT_ALLOWED", "NOT_FOUND", "MCP_TOOL_ERROR"): + repairs += 1 + if repairs > _MAX_REPAIRS: + # give up on the RUN, keep the session drivable: the + # error is recorded and a human can re-steer or finish + await self.broker.set_status( + session_id, "open", {"error": f"engine gave up: {outcome}"}) + return EngineResult("finished", {"error": f"engine gave up: {outcome}"}) + continue # the failed call is in the transcript; model self-corrects + if outcome == "AGENT_BUDGET_EXCEEDED": + await self.broker.set_status(session_id, "budget_exhausted", + {"error": out.get("error")}) + return EngineResult("budget_exhausted", {"error": out.get("error")}) + if outcome == "pending_approval": + await self.broker.set_status( + session_id, "blocked_on_approval", + {"blocked_on": out.get("approval_id"), + "message": "write is awaiting human approval; approve, then run again"}) + return EngineResult("blocked_on_approval", + {"approval_id": out.get("approval_id")}) + if outcome == "AGENT_SESSION_CLOSED": Review Comment: Good catch. The engine's outcome handling should whitelist-ok outcomes and terminate on `AGENT_SESSION_EXPIRED`. I'll fix the engine loop and add a hard step cap as a belt. ########## ontogeny/server/ontogeny/agent/broker.py: ########## @@ -0,0 +1,727 @@ +# 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. +"""AgentBroker: session lifecycle + governed tool execution. + +Every external agent call funnels through :meth:`call_tool`. The method is +the contract: authenticate the session, check the catalog, execute through +ServiceContext, record the step. Writes additionally go through the approval +gate (M2) -- a pending approval row is created and nothing executes until a +human consumes it. +""" +from __future__ import annotations + +import datetime as _dt +import logging +import time +from typing import Any + +from sqlalchemy import delete, or_, select, update + +from ..action.models import OutboxRow # noqa: F401 (re-exported for API layer typing) +from ..errors import NotFoundError, OOError, PolicyDeniedError, StoreError +from ..registry.compiled import CompiledOntology +from .catalog import ToolDef, build_catalog, catalog_hash, effective_tools +from .models import AgentApprovalRow, AgentSessionRow, AgentStepRow, args_digest + +log = logging.getLogger("ontogeny.agent") + + +def _utcnow() -> _dt.datetime: + return _dt.datetime.now(_dt.timezone.utc) + + +def _jsonable(value: Any) -> Any: + """Row objects carry dates/datetimes/Decimals; JSON columns refuse them.""" + if isinstance(value, (_dt.datetime, _dt.date)): + return value.isoformat() + if isinstance(value, dict): + return {k: _jsonable(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_jsonable(v) for v in value] + try: + import json as _json + _json.dumps(value) + return value + except (TypeError, ValueError): + return str(value) + + +class SessionBudgetExceeded(OOError): + code = "AGENT_BUDGET_EXCEEDED" + http_status = 429 + + +class ToolNotAllowed(OOError): + code = "AGENT_TOOL_NOT_ALLOWED" + http_status = 403 + + +class AgentServiceError(StoreError): + """Raised for broker-internal problems (bad session state etc.).""" + + +class AgentBroker: + def __init__(self, sc) -> None: + self.sc = sc + + # ------------------------------------------------------------- catalog + + def plugin(self, name: str) -> Any: + if self.sc.compiled is None: + raise StoreError("service not initialized") + plug = self.sc.compiled.agent_plugins.get(name) + if plug is None: + raise NotFoundError(f"unknown agent plugin {name!r}") + return plug + + def list_plugins(self) -> list[dict[str, Any]]: + if self.sc.compiled is None: + return [] + out = [] + for name, plug in self.sc.compiled.agent_plugins.items(): + tools = effective_tools(self.sc.compiled, plug) + out.append({ + "name": name, + "display": plug.metadata.display, + "description": plug.metadata.description, + "transport": plug.spec.transport.kind, + "engine": {"kind": plug.spec.engine.kind, "endpoint": plug.spec.engine.endpoint}, + "principal": plug.spec.principal.as_principal(), + "approval": plug.spec.approval.writes, + "budget": { + "steps": plug.spec.budget.steps, + "wall_ms": plug.spec.budget.wall_ms, + "writes_per_session": plug.spec.budget.writes_per_session, + }, + "tools": {n: t.to_meta() for n, t in tools.items()}, + }) + return out + + # ------------------------------------------------------------ sessions + + async def open_session(self, plugin_name: str, task: str, *, + budget: dict[str, Any] | None = None, + expires_at: _dt.datetime | None = None, + driver: dict[str, Any] | None = None) -> dict[str, Any]: + """Open a session. ``budget`` overrides the plugin's frozen limits per + field; 0 (or a negative) means UNLIMITED, an omitted field keeps the + plugin's value. ``expires_at`` bounds the session's usable life. + ``driver`` is the accountable WHO behind the session (the plugin + principal stays the EXECUTION identity; this records who asked for it). + """ + plug = self.plugin(plugin_name) + compiled: CompiledOntology = self.sc.compiled # type: ignore[assignment] + tools = effective_tools(compiled, plug) + budget = budget or {} + + def _limit(field: str, plugin_default: int) -> int: + v = budget.get(field) + if v is None: + return plugin_default + return max(int(v), 0) # 0 = unlimited Review Comment: You're right — `_limit` returns `max(plugin_default, caller_value)` which can only widen. The fix is to clamp the override to the plugin's own budget ceiling. I'll address in a follow-up. -- 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]
