You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hugegraph.apache.org by mi...@apache.org on 2026/10/21 09:41:00 UTC
(hugegraph-ai) 04/06: fix(data/robustness): projection keyset backfill, builder store sidecars, demo env respect, admin lockout self-heal, bounded reads, sandbox stdout isolation
This is an automated email from the ASF dual-hosted git repository.
simon824 pushed a commit to branch ontogeny
in repository https://gitbox.apache.org/repos/asf/hugegraph-ai.git
commit 4e1e130e1b0f81507049dbf0e81404ce494e0b3d
Author: simon824 <si...@users.noreply.github.com>
AuthorDate: Wed Oct 21 16:54:01 2026 +0800
fix(data/robustness): projection keyset backfill, builder store sidecars, demo env respect, admin lockout self-heal, bounded reads, sandbox stdout isolation
- projection rebuild: keyset pagination by PK replaces the silent
10k-row truncation without ORDER BY (edges used to reference vertices
that were never written)
- builder: inline backing.store_spec sidecars are actually written to
stores/<name>.yaml now (the writer re-readed the already-parsed spec,
so saving an object with a new backing store failed validation)
- cli --demo: operator-preset env wins (setdefault) -- demo paths used
to clobber ONTOGENY_DB_DSN and lose edits outside the /data volume
- auth: demoting/disabling the last ACTIVE administrator is refused
(counted excluding the row itself); ensure_admin self-heals an
existing-but-demoted bootstrap account instead of crashing at boot
- api: bounded reads -- QueryBody.limit/offset and TraverseBody
depth/ids/path carry pydantic bounds; the query service and the
broker's sessionless search clamp limits too (the MCP surface builds
args without pydantic)
- sandbox: user print() no longer corrupts the JSON protocol stream --
child binds the protocol to the real stdout at startup and redirects
user stdout to stderr (RPC included); the parent turns stray non-JSON
lines into SandboxError instead of an uncaught JSONDecodeError
10 new regression tests; full suite 582 passed
---
ontogeny/server/ontogeny/agent/broker.py | 2 +-
ontogeny/server/ontogeny/api/schemas.py | 14 +-
ontogeny/server/ontogeny/auth/service.py | 38 ++++-
ontogeny/server/ontogeny/builder_ops.py | 35 ++--
ontogeny/server/ontogeny/cli.py | 5 +-
ontogeny/server/ontogeny/engine/service.py | 3 +
ontogeny/server/ontogeny/functions/child.py | 19 ++-
ontogeny/server/ontogeny/functions/sandbox.py | 7 +-
ontogeny/server/ontogeny/projection/worker.py | 28 +++-
ontogeny/tests/test_pr380_data.py | 228 ++++++++++++++++++++++++++
10 files changed, 336 insertions(+), 43 deletions(-)
diff --git a/ontogeny/server/ontogeny/agent/broker.py b/ontogeny/server/ontogeny/agent/broker.py
index f024feed..274f4156 100644
--- a/ontogeny/server/ontogeny/agent/broker.py
+++ b/ontogeny/server/ontogeny/agent/broker.py
@@ -648,7 +648,7 @@ class AgentBroker:
if tdef.kind == "meta":
return "ok", sc.compiled.to_meta(), {}
if tdef.kind == "search":
- limit = int(args.get("limit") or 20)
+ limit = max(1, min(int(args.get("limit") or 20), 1000))
started = time.perf_counter()
try:
out = await sc.query.query(session, tdef.target, principal,
diff --git a/ontogeny/server/ontogeny/api/schemas.py b/ontogeny/server/ontogeny/api/schemas.py
index ef60ac83..def8e96b 100644
--- a/ontogeny/server/ontogeny/api/schemas.py
+++ b/ontogeny/server/ontogeny/api/schemas.py
@@ -22,7 +22,7 @@ from __future__ import annotations
from typing import Any, Literal
-from pydantic import BaseModel
+from pydantic import BaseModel, Field
# ------------------------------------------------------------------ schemas
@@ -30,8 +30,10 @@ from pydantic import BaseModel
class QueryBody(BaseModel):
filter: Any = None
sort: Any = None
- limit: int = 100
- offset: int = 0
+ # bounded: caller-supplied limits used to be uncapped, letting an
+ # unauthenticated query load entire tables into memory
+ limit: int = Field(100, ge=1, le=1000)
+ offset: int = Field(0, ge=0, le=100_000)
links: list[str] | None = None
via: str | None = None
@@ -67,9 +69,9 @@ class FunctionBody(BaseModel):
class TraverseBody(BaseModel):
start_type: str
- start_ids: list[str]
- path: list[dict[str, Any]]
- max_depth: int = 3
+ start_ids: list[str] = Field(max_length=50)
+ path: list[dict[str, Any]] = Field(max_length=6)
+ max_depth: int = Field(3, ge=1, le=6)
class EvolveProposalBody(BaseModel):
diff --git a/ontogeny/server/ontogeny/auth/service.py b/ontogeny/server/ontogeny/auth/service.py
index ff0fc1ca..fb0a9c5e 100644
--- a/ontogeny/server/ontogeny/auth/service.py
+++ b/ontogeny/server/ontogeny/auth/service.py
@@ -38,7 +38,7 @@ from typing import Any
from sqlalchemy import delete, func, select
-from ..errors import AuthenticationError, DSLValidationError, NotFoundError
+from ..errors import ConflictError, AuthenticationError, DSLValidationError, NotFoundError
from .models import SessionRow, UserRow
from .passwords import hash_password, new_token, token_fingerprint, verify_password
@@ -106,6 +106,7 @@ class AuthService:
row = await self._user(session, username)
if row is None:
raise NotFoundError(f"user {username!r} not found")
+ was_active_admin = row.is_admin and row.status == "active"
if "display" in patch and patch["display"] is not None:
row.display = str(patch["display"])
if patch.get("roles") is not None:
@@ -120,6 +121,21 @@ class AuthService:
row.status = patch["status"]
if patch.get("is_admin") is not None:
row.is_admin = bool(patch["is_admin"])
+ # demoting or disabling the LAST active administrator locks the
+ # deployment out after the next restart; refuse it outright
+ is_still_active_admin = row.is_admin and row.status == "active"
+ if was_active_admin and not is_still_active_admin:
+ other_active_admins = (await session.execute(
+ select(func.count()).select_from(UserRow).where(
+ UserRow.is_admin.is_(True), UserRow.status == "active",
+ UserRow.id != row.id,
+ )
+ )).scalar_one()
+ if other_active_admins == 0:
+ raise ConflictError(
+ f"cannot demote or disable {username!r}: it is the last active administrator",
+ details={"username": username},
+ )
if patch.get("password"):
self._check_password(patch["password"])
row.password_hash = hash_password(patch["password"])
@@ -240,13 +256,31 @@ class AuthService:
"""
async with self.sc.sessionmaker() as session:
existing = (await session.execute(
- select(func.count()).select_from(UserRow).where(UserRow.is_admin.is_(True))
+ select(func.count()).select_from(UserRow).where(
+ UserRow.is_admin.is_(True), UserRow.status == "active"
+ )
)).scalar_one()
if existing:
return None
username = os.environ.get("ONTOGENY_ADMIN_USER", "admin")
password = os.environ.get("ONTOGENY_ADMIN_PASSWORD") or "ontogeny@2026"
+ # self-heal instead of crashing: the bootstrap account may already
+ # exist as non-admin or disabled (demoted by a buggy older path), and
+ # create_user would raise "username exists" on the next restart
+ async with self.sc.sessionmaker() as session:
+ row = await self._user(session, username)
+ if row is not None:
+ row.is_admin = True
+ row.status = "active"
+ row.roles = sorted(set(row.roles) | {"admin"})
+ await session.commit()
+ log.warning(
+ "re-activated the bootstrap administrator %r (it existed but was "
+ "not an active admin); set ONTOGENY_ADMIN_PASSWORD and change it",
+ username,
+ )
+ return self._view(row)
user = await self.create_user(
username=username, password=password, display="平台管理员",
roles=["admin"], is_admin=True,
diff --git a/ontogeny/server/ontogeny/builder_ops.py b/ontogeny/server/ontogeny/builder_ops.py
index a656c0ad..6c0af5c7 100644
--- a/ontogeny/server/ontogeny/builder_ops.py
+++ b/ontogeny/server/ontogeny/builder_ops.py
@@ -107,34 +107,31 @@ async def builder_save(sc, resources: list[dict[str, Any]], *, deletes: list[str
import yaml as _yaml_mod
# ObjectType store sidecars: the console's data-mount editor declares the
- # source INLINE as backing.store_spec ({kind, connection}) because a bare
- # store NAME would be a placeholder, not a source. Materialize it as
- # stores/<store>.yaml (same pattern as PolicySet cedar sidecars) and strip
- # it from the object YAML -- BackingSpec has no such field on disk.
- for res in parsed:
- if res.kind != "ObjectType":
- continue
- backing = (res.spec.get("backing") if isinstance(res.spec, dict) else None) or {}
- store_spec = backing.get("store_spec")
- if not isinstance(store_spec, dict):
- continue
- store_name = str(backing.get("store") or "").strip()
- kind = str(store_spec.get("kind") or "").strip()
- connection = str(store_spec.get("connection") or "").strip()
+ # source INLINE as backing.store_spec ({kind, connection}); the first pass
+ # (above) strips it before the pydantic parse because BackingSpec has no
+ # such field on disk. THIS pass materializes each collected sidecar as
+ # stores/<name>.yaml -- it used to re-read res.spec (already parsed, never
+ # a dict), so the files were never written and publishing an object with a
+ # new backing store failed validation on the missing store.
+ for store_name, kind, connection in store_sidecars:
+ import yaml as _yaml
+
+ from ontogeny.errors import DSLValidationError
+
if not store_name or not kind or not connection:
raise DSLValidationError(
- f"ObjectType/{res.metadata.name}: incomplete data source — "
- "store name, source type and connection are all required",
+ "ObjectType backing.store_spec: store name, source type and "
+ "connection are all required",
details={"store": store_name, "kind": kind},
)
- access = "read-only" if kind == "csv" else "read-write"
stores_dir = pkg_dir / "stores"
stores_dir.mkdir(parents=True, exist_ok=True)
(stores_dir / f"{store_name}.yaml").write_text(
- _yaml_mod.safe_dump({
+ _yaml.safe_dump({
"apiVersion": "ontogeny/v1", "kind": "Store",
"metadata": {"name": store_name},
- "spec": {"type": kind, "connection": connection, "access": access},
+ "spec": {"type": kind, "connection": connection,
+ "access": "read-only" if kind == "csv" else "read-write"},
}, allow_unicode=True, sort_keys=False),
encoding="utf-8",
)
diff --git a/ontogeny/server/ontogeny/cli.py b/ontogeny/server/ontogeny/cli.py
index 26e65e6b..4b1392e3 100644
--- a/ontogeny/server/ontogeny/cli.py
+++ b/ontogeny/server/ontogeny/cli.py
@@ -253,8 +253,11 @@ def serve(
paths = prepare_demo(data_dir, package=package or DEFAULT_PACKAGE,
reseed=reseed, direct=direct)
+ # operator-provided env wins: --demo used to clobber ONTOGENY_DB_DSN
+ # (and friends), so edits/promotions landed outside the /data volume
+ # and vanished on container recreation
for k, v in paths.as_env().items():
- os.environ[k] = v
+ os.environ.setdefault(k, v)
typer.secho(f"demo data dir: {paths.root.resolve()}", fg=typer.colors.CYAN)
if direct:
typer.secho(f" serving the domain IN PLACE: {paths.package_root}\n"
diff --git a/ontogeny/server/ontogeny/engine/service.py b/ontogeny/server/ontogeny/engine/service.py
index aa0fd2f3..a330cc79 100644
--- a/ontogeny/server/ontogeny/engine/service.py
+++ b/ontogeny/server/ontogeny/engine/service.py
@@ -83,6 +83,9 @@ class QueryService:
links: list[str] | None = None,
) -> dict[str, Any]:
started = time.perf_counter()
+ # belt over the schema bounds: the MCP surface builds args without
+ # pydantic, so clamp here too
+ limit = max(1, min(int(limit or 100), 1000))
rows, total = await self.repo.query(
session, object_type, filt=filt, sort=sort, limit=limit, offset=offset,
)
diff --git a/ontogeny/server/ontogeny/functions/child.py b/ontogeny/server/ontogeny/functions/child.py
index 9d1c4f05..36ace4f7 100644
--- a/ontogeny/server/ontogeny/functions/child.py
+++ b/ontogeny/server/ontogeny/functions/child.py
@@ -30,6 +30,7 @@ Protocol (one JSON object per line):
"""
from __future__ import annotations
+import contextlib
import importlib.util
import json
import sys
@@ -45,8 +46,8 @@ class _OO:
def _rpc(self, method: str, params: dict):
self._id += 1
- sys.stdout.write(json.dumps({"type": "rpc", "id": self._id, "method": method, "params": params}) + "\n")
- sys.stdout.flush()
+ _PROTO.write(json.dumps({"type": "rpc", "id": self._id, "method": method, "params": params}) + "\n")
+ _PROTO.flush()
line = sys.stdin.readline()
if not line:
raise RuntimeError("sandbox parent closed the pipe")
@@ -87,9 +88,12 @@ def _load(module_path: str, func_name: str):
return mod, fn, None
+_PROTO = sys.stdout # the protocol stream, bound BEFORE any user-code redirect
+
+
def _emit(payload: dict) -> None:
- sys.stdout.write(json.dumps(payload, default=str) + "\n")
- sys.stdout.flush()
+ _PROTO.write(json.dumps(payload, default=str) + "\n")
+ _PROTO.flush()
def main() -> None:
@@ -103,7 +107,12 @@ def main() -> None:
try:
mode = init.get("type", "invoke")
if mode == "invoke":
- _emit({"type": "result", "value": fn(**(init.get("params") or {}))})
+ # user stdout must never touch the protocol stream (a single
+ # print() used to corrupt the JSON-RPC framing and 500 the whole
+ # call); the parent additionally treats stray non-JSON lines as
+ # SandboxError
+ with contextlib.redirect_stdout(sys.stderr):
+ _emit({"type": "result", "value": fn(**(init.get("params") or {}))})
elif mode == "map":
# derivation batch: the runner loops per row inside this process
# (one spawn for the whole batch); ontogeny.query stays available per row
diff --git a/ontogeny/server/ontogeny/functions/sandbox.py b/ontogeny/server/ontogeny/functions/sandbox.py
index 347d9b50..309592e2 100644
--- a/ontogeny/server/ontogeny/functions/sandbox.py
+++ b/ontogeny/server/ontogeny/functions/sandbox.py
@@ -242,7 +242,12 @@ class FunctionSandbox:
line = await proc.stdout.readline()
if not line:
raise SandboxError("sandbox exited before returning a result")
- msg = json.loads(line)
+ try:
+ msg = json.loads(line)
+ except json.JSONDecodeError as exc:
+ raise SandboxError(
+ f"protocol stream corrupted by non-JSON output: {line[:200]!r}"
+ ) from exc
if msg.get("type") == "rpc":
method = msg.get("method")
if method == "llm":
diff --git a/ontogeny/server/ontogeny/projection/worker.py b/ontogeny/server/ontogeny/projection/worker.py
index 6b3893e9..9a70f85c 100644
--- a/ontogeny/server/ontogeny/projection/worker.py
+++ b/ontogeny/server/ontogeny/projection/worker.py
@@ -177,14 +177,26 @@ class ProjectionWorker:
counts = {"vertices": 0, "edges": 0}
for object_type in self.projection.spec.include.objects:
props_whitelist, pk = self._whitelist(object_type)
- rows, total = await repo.query(session, object_type, limit=10000)
- items = []
- for row in rows:
- item = {p: row.get(p) for p in props_whitelist}
- item[pk] = row.get(pk)
- items.append(item)
- await self.client.upsert_vertices(object_type, pk, items)
- counts["vertices"] += len(items)
+ # keyset pagination by PK: the old `limit=10000` without ORDER BY
+ # silently truncated every larger table and left edges pointing
+ # at vertices that were never written
+ batch = 500
+ last = None
+ while True:
+ filt = {"field": pk, "op": "gt", "value": last} if last is not None else None
+ rows, _total = await repo.query(
+ session, object_type, filt=filt, sort=[[pk, "asc"]], limit=batch,
+ )
+ if not rows:
+ break
+ items = []
+ for row in rows:
+ item = {p: row.get(p) for p in props_whitelist}
+ item[pk] = row.get(pk)
+ items.append(item)
+ await self.client.upsert_vertices(object_type, pk, items)
+ counts["vertices"] += len(items)
+ last = rows[-1].get(pk)
for link_name in self.projection.spec.include.links:
lnk = self.compiled.links[link_name]
pairs = await self._link_pairs(repo, session, link_name)
diff --git a/ontogeny/tests/test_pr380_data.py b/ontogeny/tests/test_pr380_data.py
new file mode 100644
index 00000000..2bfb9b47
--- /dev/null
+++ b/ontogeny/tests/test_pr380_data.py
@@ -0,0 +1,228 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you 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.
+"""PR #380 data/robustness-pack regression tests: projection backfill
+pagination, builder store sidecars, demo env respect, admin lockout
+self-heal, bounded reads, sandbox stdout isolation."""
+from __future__ import annotations
+
+import json
+
+import pytest
+
+
+class TestProjectionKeysetBackfill:
+ async def test_backfill_pages_past_batch_size(self, client):
+ """A 25-row table with batch forced to 10 must fully land in the graph."""
+ import datetime as dt
+
+ from ontogeny.projection.worker import ProjectionWorker
+
+ c, sc = client
+ await c.post("/api/v1/admin/sync/work-center")
+ # 4 seeded + 21 fresh rows = 25 > any single page
+ for i in range(21):
+ async with sc.sessionmaker() as s:
+ await sc.repo.action_insert(
+ s, "work-center",
+ {"work_center_id": f"WC-X{i:02d}", "name": f"Extra {i}",
+ "process": "assembly", "site": "plant-north",
+ "capacity_per_shift": 10, "shifts_per_day": 1, "status": "ACTIVE"},
+ dt.datetime.now(dt.timezone.utc),
+ )
+ await s.commit()
+
+ prj = next(iter(sc.compiled.projections.values()))
+ upserted: list[int] = []
+
+ class FakeClient:
+ async def upsert_vertices(self, label, pk, items):
+ upserted.append(len(items))
+
+ async def upsert_edges(self, *a, **k):
+ return None
+
+ worker = ProjectionWorker.__new__(ProjectionWorker)
+ worker.projection = prj
+ worker.compiled = sc.compiled
+ worker.client = FakeClient()
+ worker._link_pairs = ProjectionWorker._link_pairs.__get__(worker)
+ # 强制小批量：monkeypatch 掉常量
+ import ontogeny.projection.worker as w
+
+ original_query = sc.repo.query
+
+ async def small_query(session, object_type, **kw):
+ kw["limit"] = min(kw.get("limit", 10), 10)
+ return await original_query(session, object_type, **kw)
+
+ sc.repo.query = small_query
+ try:
+ async with sc.sessionmaker() as s:
+ expected = len(await sc.repo.alive_ids(s, "work-center"))
+ counts = await worker.rebuild(sc.repo, s)
+ finally:
+ sc.repo.query = original_query
+ assert counts["vertices"] == expected, "backfill must page through every row"
+ assert all(n <= 10 for n in upserted), "each page respects the batch bound"
+
+
+class TestBuilderStoreSidecars:
+ async def test_inline_store_spec_materializes_store_yaml(self, client):
+ c, sc = client
+ obj = {
+ "apiVersion": "ontogeny/v1", "kind": "ObjectType",
+ "metadata": {"name": "widget", "display": "Widget"},
+ "spec": {
+ "primaryKey": ["widget_id"],
+ "properties": {
+ "widget_id": {"type": "string", "display": "ID", "required": True},
+ "name": {"type": "string", "display": "Name"},
+ },
+ "backing": {
+ "store": "widget-src",
+ "mode": "materialized",
+ "source": {"table": "widgets"},
+ "store_spec": {"kind": "sqlite",
+ "connection": "sqlite:///widgets.db"},
+ },
+ },
+ }
+ r = await c.post("/api/v1/admin/builder/save", json={"resources": [obj]})
+ assert r.status_code == 200, r.text
+ import pathlib
+
+ store_yaml = pathlib.Path(sc.package_root) / "stores" / "widget-src.yaml"
+ assert store_yaml.is_file(), "the inline store sidecar must be written"
+ assert "sqlite" in store_yaml.read_text(encoding="utf-8")
+
+
+class TestDemoEnvRespect:
+ def test_as_env_does_not_clobber_operator_values(self, tmp_path, monkeypatch):
+ from ontogeny.demo.bootstrap import prepare_demo
+
+ monkeypatch.setenv("ONTOGENY_DB_DSN", "sqlite+aiosqlite:////data/operator.db")
+ paths = prepare_demo(tmp_path / "demo", package="domains/product-manufacturing")
+ env = paths.as_env()
+ # as_env itself only carries defaults; the respect happens in cli via
+ # setdefault -- assert the contract here and simulate the cli line
+ for k, v in env.items():
+ os_setdefault = {k: v}
+ import os
+
+ assert os.environ["ONTOGENY_DB_DSN"].endswith("/operator.db")
+ # the exact cli behaviour under test:
+ for k, v in env.items():
+ os.environ.setdefault(k, v)
+ assert os.environ["ONTOGENY_DB_DSN"].endswith("/operator.db"), \
+ "operator-preset env must survive a --demo boot"
+
+
+class TestAdminLockout:
+ async def test_cannot_demote_last_admin(self, client):
+ from ontogeny.errors import ConflictError
+
+ c, sc = client
+ with pytest.raises(ConflictError):
+ await sc.auth.update_user("admin", {"is_admin": False})
+ with pytest.raises(ConflictError):
+ await sc.auth.update_user("admin", {"status": "disabled"})
+
+ async def test_second_admin_can_be_demoted(self, client):
+ c, sc = client
+ await sc.auth.create_user(username="admin2", password="password123",
+ roles=["admin"], is_admin=True)
+ out = await sc.auth.update_user("admin2", {"is_admin": False})
+ assert out["is_admin"] is False
+
+ async def test_ensure_admin_self_heals_disabled_bootstrap(self, client):
+ c, sc = client
+ # 直接把 root 停用（绕过守卫模拟历史脏数据），ensure_admin 应复活它而非崩溃
+ from sqlalchemy import update
+
+ from ontogeny.auth.models import UserRow
+
+ async with sc.sessionmaker() as s:
+ await s.execute(update(UserRow).values(status="disabled"))
+ await s.commit()
+ user = await sc.auth.ensure_admin()
+ assert user is not None and user["is_admin"] is True and user["status"] == "active"
+
+
+class TestBoundedReads:
+ async def test_query_limit_rejected_over_bound(self, client):
+ c, _ = client
+ r = await c.post("/api/v1/objects/work-center/query", json={"limit": 100000})
+ assert r.status_code == 422
+
+ async def test_traverse_depth_rejected_over_bound(self, client):
+ c, _ = client
+ r = await c.post("/api/v1/graph/traverse", json={
+ "start_type": "production-order", "start_ids": ["PO-1001"],
+ "path": [], "max_depth": 10,
+ })
+ assert r.status_code == 422
+
+ async def test_readonly_tool_limit_clamped(self, client):
+ c, sc = client
+ await c.post("/api/v1/admin/sync/work-center")
+ out = await sc.agents.read_only_tool("search_work_center", None, {"limit": 999999})
+ assert out["outcome"] == "ok" and len(out["detail"]["objects"]) <= 1000
+
+
+class TestSandboxStdoutIsolation:
+ async def test_print_does_not_break_protocol(self, tmp_path, golden_pkg_path):
+ import shutil
+
+ pkg = tmp_path / "pkg"
+ shutil.copytree(golden_pkg_path, pkg)
+ (pkg / "functions" / "chatty.py").write_text(
+ "def say(what):\n"
+ " print('user-side noise:', what)\n"
+ " return {'said': what}\n",
+ encoding="utf-8",
+ )
+ import yaml as _yaml
+
+ doc = _yaml.safe_load((pkg / "functions" / "capacity-check.yaml").read_text(encoding="utf-8"))
+ doc["metadata"]["name"] = "chatty"
+ doc["spec"]["entry"] = "chatty.py:say"
+ doc["spec"]["parameters"] = {"what": {"type": "string", "display": "What", "required": True}}
+ doc["spec"]["capabilities"] = []
+ doc["spec"].pop("returns", None)
+ (pkg / "functions" / "chatty.yaml").write_text(
+ _yaml.safe_dump(doc, allow_unicode=True, sort_keys=False), encoding="utf-8")
+
+ from ontogeny.config import Settings
+ from ontogeny.core import load_package
+ from ontogeny.db import init_schema, make_engine, make_sessionmaker
+ from ontogeny.functions.sandbox import FunctionSandbox
+ from ontogeny.registry.store import RegistryService
+
+ settings = Settings(env={"ERP_DSN": "sqlite+aiosqlite:///:memory:"})
+ engine = make_engine("sqlite+aiosqlite:///:memory:")
+ await init_schema(engine)
+ maker = make_sessionmaker(engine)
+ async with maker() as s:
+ compiled = await RegistryService().publish(s, load_package(str(pkg)))
+ await s.commit()
+ fn = compiled.functions["chatty"]
+ from ontogeny.stores import ObjectRepository
+
+ sandbox = FunctionSandbox(compiled, ObjectRepository(compiled), timeout_s=15)
+ async with maker() as s:
+ value = await sandbox.run(s, fn, {"what": "hello"})
+ assert value == {"said": "hello"}