simon824 commented on code in PR #380: URL: https://github.com/apache/hugegraph-ai/pull/380#discussion_r4336332160
########## server/ontogeny/cli.py: ########## @@ -0,0 +1,343 @@ +# 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. +"""ontogeny -- the operator's door. Everything it does goes through +ServiceContext (same governance as HTTP/MCP) or ontogeny.core (pure validation).""" +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +from pathlib import Path + +import typer + +from .config import load_settings + +app = typer.Typer(no_args_is_help=True, help="ontogeny control plane CLI") +evolve_app = typer.Typer(no_args_is_help=True, help="RSI evolution loop") +projection_app = typer.Typer(no_args_is_help=True, help="graph projection") +app.add_typer(evolve_app, name="evolve") +app.add_typer(projection_app, name="projection") + + +# ----------------------------------------------------------------- pure cmds + + [email protected]() +def validate(path: str) -> None: + """Validate a package (structure + semantics). Exit 1 on errors.""" + from .core import load_package, validate + from .errors import DSLValidationError + + try: + pkg = load_package(path) + except DSLValidationError as exc: + typer.secho(f"FAIL load: {exc.message}", fg=typer.colors.RED, err=True) + raise typer.Exit(1) + rep = validate(pkg) + for i in rep.issues: + color = typer.colors.RED if i.severity == "error" else typer.colors.YELLOW + typer.secho(f"{i.severity.upper():7} {i.code:22} {i.resource}: {i.message}", fg=color) + if rep.ok: + typer.secho(f"OK ({rep.summary()})", fg=typer.colors.GREEN) + else: + raise typer.Exit(1) + + [email protected]() +def lint(path: str) -> None: + from .core import load_package + from .core.linter import lint as run_lint + + rep = run_lint(load_package(path)) + for i in rep.issues: + typer.secho(f"WARN {i.code:8} {i.resource}: {i.message}", fg=typer.colors.YELLOW) + typer.secho(f"lint done ({rep.summary()})", fg=typer.colors.GREEN) + + [email protected]() +def migrate(db: str = None, apply: bool = False) -> None: + """Diff the DSL (compiled snapshot) against the live object tables. + + Additive changes are applied automatically by ``ensure_tables`` on every + start; this command surfaces everything BEYOND that: type changes, dropped + or renamed properties, and orphaned tables a rename or delete left behind. + ``--apply`` runs only the safe additive class -- destructive steps are + printed for a human, never executed, because the object tables are the + authoritative data (docs/data-layer.md: "真源（数据）"). + """ + from .service import ServiceContext + from .stores.migrate import apply_safe, plan_migration + + settings = load_settings(db_dsn=db) if db else load_settings() + sc = ServiceContext(settings, os.environ.get("ONTOGENY_PACKAGE_ROOT")) + + async def run() -> int: + await sc.initialize() + async with sc.sessionmaker() as s: + conn = await s.connection() + plan = await conn.run_sync(lambda c: plan_migration(c, sc.compiled)) + if apply: + n = await conn.run_sync(lambda c: apply_safe(c, plan)) + await s.commit() + return n + typer.echo(plan.render()) + if plan.destructive: + typer.secho( + f"\n{len(plan.destructive)} destructive step(s) require manual review " + "(the object tables are authoritative data)", + fg=typer.colors.YELLOW, + ) + return len(plan.safe) + + n = asyncio.run(run()) + typer.secho(f"migration: {n} safe step(s){' applied' if apply else ' pending'}", + fg=typer.colors.GREEN) + + + [email protected]() +def docs(path: str, out: str = "docs-generated") -> None: + """Generate a human-readable markdown view of the package.""" + from .core import load_package + + pkg = load_package(path) + outdir = Path(out) + outdir.mkdir(parents=True, exist_ok=True) + lines = [f"# Ontology: {pkg.manifest.metadata.name}", ""] + for r in pkg.resources: + if r.kind == "Ontology": + continue + meta = r.metadata + lines.append(f"## {r.kind}: {meta.name}") + lines.append(f"- display: {meta.display or '-'}") + if meta.description: + lines.append(f"- description: {meta.description}") + lines.append("") + (outdir / "ontology.md").write_text("\n".join(lines), encoding="utf-8") + typer.secho(f"docs written to {outdir/'ontology.md'}", fg=typer.colors.GREEN) + + [email protected]() +def diff(a: str, b: str) -> None: + """Two-package semantic diff (names added/removed/changed).""" + from .core import load_package + + pa, pb = load_package(a), load_package(b) + sa = {(r.kind, r.metadata.name): r.model_dump_json() for r in pa.resources} + sb = {(r.kind, r.metadata.name): r.model_dump_json() for r in pb.resources} + for key in sorted(set(sa) | set(sb)): + if key not in sb: + typer.secho(f"- {key[0]}/{key[1]}", fg=typer.colors.RED) + elif key not in sa: + typer.secho(f"+ {key[0]}/{key[1]}", fg=typer.colors.GREEN) + elif sa[key] != sb[key]: + typer.secho(f"~ {key[0]}/{key[1]}", fg=typer.colors.YELLOW) + + +# ------------------------------------------------------------ service cmds + + [email protected]() +def publish(path: str = None, db: str = None) -> None: + """Validate + publish a package into the registry.""" + path = path or os.environ.get("ONTOGENY_PACKAGE_ROOT") + if not path: + typer.secho("path or ONTOGENY_PACKAGE_ROOT required", fg=typer.colors.RED, err=True) + raise typer.Exit(2) + from .service import ServiceContext + + settings = load_settings(db_dsn=db) if db else load_settings() + sc = ServiceContext(settings, None) + + async def run(): + await sc.initialize() + compiled = await sc.publish(path) + return compiled.content_hash + + h = asyncio.run(run()) + typer.secho(f"published content_hash={h[:12]}", fg=typer.colors.GREEN) + + [email protected]() +def sync(object_type: str, db: str = None) -> None: + from .service import ServiceContext + + settings = load_settings(db_dsn=db) if db else load_settings() + sc = ServiceContext(settings, os.environ.get("ONTOGENY_PACKAGE_ROOT")) + + async def run(): + await sc.initialize() + return await sc.sync(object_type) + + typer.echo(json.dumps(asyncio.run(run()))) + + +def _repo_root() -> Path: + return Path(__file__).resolve().parents[2] + + +def _ensure_ui(build: bool, ui_dir: str | None) -> Path | None: + """Locate the built SPA, optionally building it (npm) when missing.""" + from .api.app import resolve_ui_dir + + resolved = resolve_ui_dir(ui_dir) + if resolved is not None or not build: + return resolved + + web = _repo_root() / "web" + dist = web / "dist" + if not (web / "package.json").is_file(): + typer.secho(f"no web/ directory at {web}; serving API only", fg=typer.colors.YELLOW) + return None + if not (web / "node_modules").is_dir(): + typer.secho( + "frontend dependencies missing -- run `npm install` in web/ first, or use --no-build-ui", + fg=typer.colors.YELLOW, + ) + return None + + typer.secho("building frontend (web/dist)…", fg=typer.colors.CYAN) + result = subprocess.run(["npm", "run", "build"], cwd=web, capture_output=True, text=True) + if result.returncode != 0: + typer.secho(f"frontend build failed:\n{result.stderr[-1500:]}", fg=typer.colors.RED) + return None + typer.secho("frontend built", fg=typer.colors.GREEN) + return resolve_ui_dir(dist) + + [email protected]() +def serve( + db: str = None, + host: str = "127.0.0.1", + port: int = 8000, + demo: bool = typer.Option(False, "--demo", help="self-contained demo: seed source data, copy the example package, serve the UI"), + data_dir: str = typer.Option(".ontogeny-demo", "--data-dir", help="demo state directory (metadata + source db + package copy)"), + package: str = typer.Option(None, "--package", help="ontology package to install into the demo (default: domains/product-manufacturing)"), + reseed: bool = typer.Option(False, "--reseed", help="rebuild the demo source db and package copy from scratch"), + build_ui: bool = typer.Option(True, "--build-ui/--no-build-ui", help="build web/dist when missing"), + ui_dir: str = typer.Option(None, "--ui-dir", help="serve an already-built SPA from this directory"), + open_browser: bool = typer.Option(False, "--open", help="open the UI in a browser once ready"), +) -> None: + """Run the HTTP API (and the built UI on the same port when available).""" + import uvicorn + + if demo: + from .demo import DEFAULT_PACKAGE, prepare_demo + + # A first-class domain (a package directly inside a domains/ root) is + # served IN PLACE: console saves — model edits and the canvas + # layout.yaml — land in domains/<name>/ instead of a throwaway copy + # that the demo's runtime reset would later wipe. Anything else (an + # ad-hoc directory, tests) keeps the copy + pristine behavior. + pkg_arg = Path(package or DEFAULT_PACKAGE) + if not pkg_arg.is_absolute(): + pkg_arg = Path.cwd() / pkg_arg + direct = pkg_arg.resolve().parent.name == "domains" + + paths = prepare_demo(data_dir, package=package or DEFAULT_PACKAGE, + reseed=reseed, direct=direct) + for k, v in paths.as_env().items(): + os.environ[k] = v Review Comment: Fixed: `as_env()` values are applied via `os.environ.setdefault()` — operator-preset env always wins. Commit 4e1e130. ########## web/src/pages/ActionRunner.tsx: ########## @@ -0,0 +1,208 @@ +/* + * 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. + */ +import { useState } from 'react' +import { Link, useParams } from 'react-router-dom' +import { useMutation, useQueryClient } from '@tanstack/react-query' +import { apiClient } from '../api/client' +import { useMeta } from '../api/queries' +import { useI18n } from '../i18n' +import { Badge, Button, Card, KeyValue, Page, Skeleton } from '../components/ui' +import { ErrorBanner } from '../components/ErrorBanner' +import { JsonView } from '../components/JsonView' +import { ParamForm, type ParamValues } from '../components/ParamForm' +import { IconArrowLeft, IconCheck, IconPlay, IconX } from '../components/icons' + +interface ValidateResult { + parameters: Record<string, unknown> + rules: Array<{ expr: string; ok: boolean; message: string | null }> + policy: { allow: boolean; reason: string } +} +interface ExecuteResult { + revision_id: number + outcome: string + object_id: string + after: Record<string, unknown> | null +} + +export function ActionRunner() { + const { name = '' } = useParams() + const { t } = useI18n() + const qc = useQueryClient() + const meta = useMeta() + const action = meta.data?.actions[name] + + const [values, setValues] = useState<ParamValues>({}) + const [targetId, setTargetId] = useState('') + const [idemKey, setIdemKey] = useState('') + const [expectedRev, setExpectedRev] = useState('') + + const validate = useMutation({ + mutationFn: () => apiClient.validateAction(name, { parameters: values, target_id: targetId || null }), + }) + const execute = useMutation({ + mutationFn: () => + apiClient.executeAction(name, { + parameters: values, + target_id: targetId || null, + expected_revision: expectedRev ? Number(expectedRev) : null, + idempotency_key: idemKey || null, + }), + // an execute writes the target object, its revision history and the audit + // trail -- invalidate those, not the whole cache (the unscoped call threw + // away the loaded meta, extensions and projection too) + onSuccess: () => { + qc.invalidateQueries({ queryKey: ['object'] }) + qc.invalidateQueries({ queryKey: ['history'] }) + qc.invalidateQueries({ queryKey: ['revisions'] }) + qc.invalidateQueries({ queryKey: ['objects'] }) + qc.invalidateQueries({ queryKey: ['data-preview'] }) + }, + }) + + if (meta.isLoading) return <Skeleton rows={6} /> + if (!action) return <ErrorBanner error={{ code: 'NOT_FOUND', message: `unknown action: ${name}` }} /> + + const missing = Object.entries(action.parameters) + .filter(([n, p]) => p.required && (values[n] === undefined || values[n] === null || values[n] === '')) + .map(([n]) => n) + + return ( + <Page> + <h1 className="sr-only">{name}</h1> + + <div className="grid gap-5 lg:grid-cols-[1.15fr_1fr]"> + {/* breadcrumb, action name and target chip live on the parameters + panel: it is the page's one primary surface */} + <Card + title={ + <span className="flex items-center gap-2"> + <Link className="ontogeny-link inline-flex items-center gap-1 text-[12.5px] muted" to={`/ontology/${action.target}`}> + <IconArrowLeft width={13} height={13} /> + {action.target} + </Link> + <span className="muted">/</span> + <span className="font-mono">{name}</span> + </span> + } + description={action.display} + actions={<Badge tone="violet">{t('action.target')}: {action.target}</Badge>} + > + <ParamForm parameters={action.parameters} value={values} onChange={setValues} /> + <div className="mt-4 grid gap-3 sm:grid-cols-3"> + <div> + <label className="ontogeny-label" htmlFor="target-id">{t('action.targetId')}</label> + <input id="target-id" className="ontogeny-input font-mono" value={targetId} onChange={(e) => setTargetId(e.target.value)} placeholder={action.target} /> + </div> + <div> + <label className="ontogeny-label" htmlFor="idem">{t('action.idempotency')}</label> + <input id="idem" className="ontogeny-input font-mono" value={idemKey} onChange={(e) => setIdemKey(e.target.value)} placeholder={t('action.optional')} /> + </div> + <div> + <label className="ontogeny-label" htmlFor="rev">{t('action.expectedRev')}</label> + <input id="rev" className="ontogeny-input font-mono" value={expectedRev} onChange={(e) => setExpectedRev(e.target.value)} placeholder="1" /> + </div> + </div> + <div className="mt-4 flex flex-wrap gap-2"> + <Button onClick={() => validate.mutate()} disabled={missing.length > 0} data-testid="validate-action" icon={<IconCheck width={14} height={14} />}> + {t('action.validate')} + </Button> + <Button variant="primary" onClick={() => execute.mutate()} disabled={missing.length > 0} data-testid="execute-action" icon={<IconPlay width={14} height={14} />}> Review Comment: Fixed: Execute button disables on `execute.isPending`; an auto idempotency key is minted per intent (double-click shares the key, the server's unique index dedupes). Commit c70119a. ########## ontogeny/web/src/pages/ActionRunner.tsx: ########## @@ -0,0 +1,208 @@ +/* + * 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. + */ +import { useState } from 'react' +import { Link, useParams } from 'react-router-dom' +import { useMutation, useQueryClient } from '@tanstack/react-query' +import { apiClient } from '../api/client' +import { useMeta } from '../api/queries' +import { useI18n } from '../i18n' +import { Badge, Button, Card, KeyValue, Page, Skeleton } from '../components/ui' +import { ErrorBanner } from '../components/ErrorBanner' +import { JsonView } from '../components/JsonView' +import { ParamForm, type ParamValues } from '../components/ParamForm' +import { IconArrowLeft, IconCheck, IconPlay, IconX } from '../components/icons' + +interface ValidateResult { + parameters: Record<string, unknown> + rules: Array<{ expr: string; ok: boolean; message: string | null }> + policy: { allow: boolean; reason: string } +} +interface ExecuteResult { + revision_id: number + outcome: string + object_id: string + after: Record<string, unknown> | null +} + +export function ActionRunner() { + const { name = '' } = useParams() + const { t } = useI18n() + const qc = useQueryClient() + const meta = useMeta() + const action = meta.data?.actions[name] + + const [values, setValues] = useState<ParamValues>({}) Review Comment: Fixed: all parameterized routes (`/objects/:type`, `/objects/:type/:id`, `/actions/:name`, `/functions/:name`) remount on param change via a `KeyedRoute` wrapper. Commit c70119a. ########## ontogeny/server/ontogeny/stores/sources.py: ########## @@ -0,0 +1,159 @@ +# 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. +"""Source stores: read connectors feeding the sync engine. + +- SqlSource: postgres/sqlite source databases via a dedicated engine +- CsvSource: snapshot-only file source + +The Store SPI is deliberately tiny: introspection + batched pull. Writes back +to source systems happen through the outbox (webhook/writeback adapters), +never through these readers. +""" +from __future__ import annotations + +import csv +from pathlib import Path +from typing import Any, AsyncIterator + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine + +from ..core.models import SourceRef, StoreResource +from ..errors import StoreError + + +def source_engine(store: StoreResource, resolved_dsn: str) -> AsyncEngine: + if store.spec.type in ("postgres", "sqlite", "mysql"): + # a bare mysql:// URL has no async driver; normalize to SQLAlchemy's + # asyncio scheme (requires aiomysql -- the sync error names it plainly) + if store.spec.type == "mysql" and resolved_dsn.startswith("mysql://"): + resolved_dsn = resolved_dsn.replace("mysql://", "mysql+aiomysql://", 1) + return create_async_engine(resolved_dsn, future=True) + raise StoreError(f"store type {store.spec.type!r} has no SQL source engine") + + +def qualified_name(ref: SourceRef, dsn: str) -> str: + if dsn.startswith("sqlite"): + return ref.table # sqlite has no schemas in the postgres sense + return f"{ref.schema_}.{ref.table}" if ref.schema_ else ref.table + + +class SqlSource: + def __init__(self, store: StoreResource, dsn: str) -> None: + self.store = store + self.dsn = dsn + self.engine: AsyncEngine | None = None + + async def _eng(self) -> AsyncEngine: + if self.engine is None: + self.engine = source_engine(self.store, self.dsn) + return self.engine + + async def columns(self, ref: SourceRef) -> set[str]: + eng = await self._eng() + async with eng.connect() as conn: + result = await conn.execute(text( + f"SELECT * FROM {qualified_name(ref, self.dsn)} LIMIT 0" + )) + return set(result.keys()) + + async def fetch( + self, ref: SourceRef, *, columns: list[str] | None = None, + watermark_col: str | None = None, after: Any = None, batch: int = 500, + ) -> AsyncIterator[dict[str, Any]]: + eng = await self._eng() + table = qualified_name(ref, self.dsn) + is_sqlite = self.dsn.startswith("sqlite") + async with eng.connect() as conn: + if watermark_col is not None: + # Keyset pagination on the watermark: loop until a short batch + # proves the `> after` range is drained. The old code returned + # after the FIRST batch, so any source change touching more + # than `batch` rows was permanently lost behind the advanced + # watermark. + cursor = after + while True: + sql = f'SELECT * FROM {table}' + params: dict[str, Any] = {} + if cursor is not None: + sql += f' WHERE "{watermark_col}" > :wm' + params["wm"] = cursor + sql += " ORDER BY rowid" if is_sqlite else f' ORDER BY "{watermark_col}"' Review Comment: Fixed: watermark fetch uses keyset pagination (`WHERE wm > :cursor ORDER BY wm LIMIT :batch`) looping until a short/empty page, so rows past the first batch are no longer lost. Commit from earlier hardening. -- 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]
