bitflicker64 commented on code in PR #391: URL: https://github.com/apache/hugegraph-ai/pull/391#discussion_r4487959870
########## ontogeny/server/ontogeny/documents/engine.py: ########## @@ -0,0 +1,1655 @@ +# 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. + +"""Document ingestion orchestration: corpus -> packets -> objects -> review. + +The engine owns the only place where agent output becomes platform data. Two +rules drive the whole module: + +* an extraction run is *replayable*: packets are persisted and re-bound without + another model call, so an ontology change is a rebind, not a re-bill; +* nothing ambiguous is written silently: missing identity, conflicting values + and unresolved edge endpoints become review candidates. +""" + +from __future__ import annotations + +import hashlib +import logging +from pathlib import Path +from typing import Any + +from sqlalchemy import delete, select + +from ..action.models import OutboxRow, RevisionRow +from ..core.models import DocumentCfg, ExtractCfg, ExtractedJoin +from ..errors import NotFoundError, OOError, StoreError +from ..stores.repo import utcnow +from . import corpus +from .binder import Binder, BoundObject, _jsonable +from .models import ( + DocumentRow, + ExtractAssertionRow, + ExtractCandidateRow, + ExtractOriginRow, + ExtractPacketRow, + ExtractRunRow, +) +from .protocol import DocInput, ExtractOptions, ExtractRequest +from .schema import compile_extraction_schema + +log = logging.getLogger("ontogeny.documents") + +PROMPT_VERSION = "ontogeny-doc-extract/v1" + + +class DocumentError(OOError): + code = "DOCUMENT_ERROR" + http_status = 422 + + +def _valid(value: Any) -> bool: + return value is not None and value != "" and value != [] and value != {} + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +class DocumentService: + def __init__(self, sc) -> None: + self.sc = sc + self._locks: dict[str, Any] = {} + from .tasks import TaskService + + self.tasks = TaskService(self) + + # --------------------------------------------------------------- helpers + + @property + def compiled(self): + return self.sc.compiled + + def _extractor(self): + return getattr(self.sc, "document_extractor", None) + + def document_stores(self) -> list[Any]: + return sorted( + (s for s in self.compiled.stores.values() if s.spec.type == "document"), + key=lambda s: s.metadata.name, + ) + + def store(self, name: str): + store = self.compiled.stores.get(name) + if store is None or store.spec.type != "document": + raise NotFoundError(f"no document store named {name!r}") + return store + + def object_types(self, store_name: str) -> list[str]: + out = [] + for name, obj in self.compiled.objects.items(): + b = obj.spec.backing + if b is not None and b.store == store_name and b.sync is not None and b.sync.strategy == "extract": + out.append(name) + return sorted(out) + + def _cfg(self, store) -> DocumentCfg: + return store.spec.documents or DocumentCfg() + + def _root(self, store) -> Path: + return corpus.resolve_root(store, self.sc.package_root) + + def _instructions(self, cfg: DocumentCfg) -> str | None: + if not cfg.prompt_file: + return None + root = self.sc.package_root + if not root: + raise DocumentError("prompt_file requires ONTOGENY_PACKAGE_ROOT") + path = (Path(root) / cfg.prompt_file).resolve() + if not str(path).startswith(str(Path(root).resolve())): + raise DocumentError("prompt_file must stay inside the package root") + if not path.is_file(): + raise DocumentError(f"prompt_file not found: {cfg.prompt_file}") + return path.read_text(encoding="utf-8") + + # ------------------------------------------------------- default source + + async def ensure_default_source(self) -> dict[str, Any]: + """Create a generic document corpus when the package has none. + + This is an explicit admin action, not a silent page-load write: adding + an object type changes the ontology, so the caller must ask for it. + """ + existing = self.document_stores() + if existing: + return {"created": False, "store": existing[0].metadata.name} + root = self.sc.package_root + if not root: + raise DocumentError("ONTOGENY_PACKAGE_ROOT is not set; cannot create a default source") + package = Path(root) + stores_dir = package / "stores" + objects_dir = package / "objects" + stores_dir.mkdir(parents=True, exist_ok=True) + objects_dir.mkdir(parents=True, exist_ok=True) + store_name = "documents" + suffix = 2 + while (stores_dir / f"{store_name}.yaml").exists() or store_name in self.compiled.stores: + store_name = f"documents-{suffix}" + suffix += 1 + if (stores_dir / f"{store_name}.yaml").exists(): + raise DocumentError(f"store resource {store_name!r} already exists") + type_name = "document-note" + suffix = 2 + while (objects_dir / f"{type_name}.yaml").exists() or type_name in self.compiled.objects: + type_name = f"document-note-{suffix}" + suffix += 1 + store_yaml = f"""apiVersion: ontogeny/v1 +kind: Store +metadata: + name: {store_name} + display: Documents + description: Default document corpus for document intelligence +spec: + type: document + connection: data/docs + documents: + include: + - "**/*.md" + - "**/*.txt" + - "**/*.pdf" + - "**/*.docx" + split: paragraph + language: en +""" + type_yaml = f"""apiVersion: ontogeny/v1 +kind: ObjectType +metadata: + name: {type_name} + display: Document note + description: A fact extracted from a document +spec: + primaryKey: [note_id] + properties: + note_id: + type: string + display: Note ID + required: true + owner: document + title: + type: string + display: Title + required: true + owner: document + summary: + type: string + display: Summary + owner: document + body: + type: string + display: Body + owner: document + backing: + store: {store_name} + mode: materialized + source: + table: document-notes + sync: + strategy: extract + extract: + key: [title] + match: [title] + create: auto + update: fill-empty +""" + (stores_dir / f"{store_name}.yaml").write_text(store_yaml, encoding="utf-8") + (objects_dir / f"{type_name}.yaml").write_text(type_yaml, encoding="utf-8") + (package / "data" / "docs").mkdir(parents=True, exist_ok=True) + await self.sc.publish(package) + return {"created": True, "store": store_name, "object_type": type_name} + + # ---------------------------------------------------------------- corpus + + async def _sync_documents( + self, session, store, cfg, scanned: list[corpus.ScannedDocument] + ) -> tuple[list[DocumentRow], list[str]]: + rows = ( + (await session.execute(select(DocumentRow).where(DocumentRow.store == store.metadata.name))).scalars().all() + ) + by_path = {str(r.path): r for r in rows if r.status == "active"} + now = utcnow() + active: list[DocumentRow] = [] + changed: list[DocumentRow] = [] + seen: set[str] = set() + for item in scanned: + key = str(item.path) + seen.add(key) + row = by_path.get(key) + if row is not None and row.id == item.doc_id: + row.last_seen_at = now + row.status = "active" + active.append(row) + changed.append(row) + continue + if row is not None and row.id != item.doc_id: + # same path, new content: the old document is removed, the new + # one is a different id (content-addressed) + row.status = "removed" + changed.append(row) + new = DocumentRow( Review Comment: Important: `by_path` only holds active rows, so a removed document that comes back with the same path and bytes (re-upload after `DELETE /documents/{doc_id}`, or a file reverted to an earlier version) inserts a second `DocumentRow` with the existing id. The `IntegrityError` then fails every `list_documents` and ingest for the store. Reactivate the existing row when `item.doc_id` already exists instead of inserting a new one. ########## ontogeny/server/ontogeny/stores/repo.py: ########## @@ -293,6 +387,50 @@ async def sync_upsert( await session.execute(update(t).where(where).values(_synced_at=ts)) return "noop" + ts = _next_version_time(current["_valid_from"], ts) + await session.execute(update(t).where(where).values(_valid_to=ts)) + merged = {c: current.get(c) for c in self._prop_cols(object_type)} + merged.update({c: current.get(c) for c in self._materialized_cols(object_type)}) + merged.update(changes) + merged.update({"_rev": current["_rev"] + 1, "_valid_from": ts, "_valid_to": None, "_synced_at": ts}) + await session.execute(t.insert().values(**merged)) + return "updated" + + async def apply_extracted( + self, + session, + object_type: str, + values: dict[str, Any], + ts: _dt.datetime, + ) -> str: + """Write document-bound values with explicit authority. + + The binder has already decided *which* fields may change (ownership, + fill-only, review); this method only makes the write versioned and + idempotent. ``None`` never overwrites an existing value: an absent + extraction is not evidence of absence. + """ + t = self._table(object_type) + pk_cols = self._pk_cols(object_type) + missing = [c for c in pk_cols if values.get(c) is None] + if missing: + raise StoreError(f"extracted row missing primary key {missing}") + where = _and(t.c._valid_to.is_(None), *[getattr(t.c, c) == values[c] for c in pk_cols]) + current = (await session.execute(select(t).where(where).limit(1).with_for_update())).mappings().first() + + if current is None: + seed = {c: values.get(c) for c in self._prop_cols(object_type) if c in values} Review Comment: Important: `apply_extracted` matches column names against a dict keyed by property names, so a property such as `riskLevel` or `mask-cost` (column `risk_level`, `mask_cost`) is dropped without a gap or candidate, and a non-snake primary key raises `missing primary key`. Translate `values` with `_to_columns` and coerce keys with `_key_values`, as `sync_upsert` does. ########## ontogeny/server/ontogeny/documents/engine.py: ########## @@ -0,0 +1,1655 @@ +# 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. + +"""Document ingestion orchestration: corpus -> packets -> objects -> review. + +The engine owns the only place where agent output becomes platform data. Two +rules drive the whole module: + +* an extraction run is *replayable*: packets are persisted and re-bound without + another model call, so an ontology change is a rebind, not a re-bill; +* nothing ambiguous is written silently: missing identity, conflicting values + and unresolved edge endpoints become review candidates. +""" + +from __future__ import annotations + +import hashlib +import logging +from pathlib import Path +from typing import Any + +from sqlalchemy import delete, select + +from ..action.models import OutboxRow, RevisionRow +from ..core.models import DocumentCfg, ExtractCfg, ExtractedJoin +from ..errors import NotFoundError, OOError, StoreError +from ..stores.repo import utcnow +from . import corpus +from .binder import Binder, BoundObject, _jsonable +from .models import ( + DocumentRow, + ExtractAssertionRow, + ExtractCandidateRow, + ExtractOriginRow, + ExtractPacketRow, + ExtractRunRow, +) +from .protocol import DocInput, ExtractOptions, ExtractRequest +from .schema import compile_extraction_schema + +log = logging.getLogger("ontogeny.documents") + +PROMPT_VERSION = "ontogeny-doc-extract/v1" + + +class DocumentError(OOError): + code = "DOCUMENT_ERROR" + http_status = 422 + + +def _valid(value: Any) -> bool: + return value is not None and value != "" and value != [] and value != {} + + +def _sha(text: str) -> str: + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +class DocumentService: + def __init__(self, sc) -> None: + self.sc = sc + self._locks: dict[str, Any] = {} + from .tasks import TaskService + + self.tasks = TaskService(self) + + # --------------------------------------------------------------- helpers + + @property + def compiled(self): + return self.sc.compiled + + def _extractor(self): + return getattr(self.sc, "document_extractor", None) + + def document_stores(self) -> list[Any]: + return sorted( + (s for s in self.compiled.stores.values() if s.spec.type == "document"), + key=lambda s: s.metadata.name, + ) + + def store(self, name: str): + store = self.compiled.stores.get(name) + if store is None or store.spec.type != "document": + raise NotFoundError(f"no document store named {name!r}") + return store + + def object_types(self, store_name: str) -> list[str]: + out = [] + for name, obj in self.compiled.objects.items(): + b = obj.spec.backing + if b is not None and b.store == store_name and b.sync is not None and b.sync.strategy == "extract": + out.append(name) + return sorted(out) + + def _cfg(self, store) -> DocumentCfg: + return store.spec.documents or DocumentCfg() + + def _root(self, store) -> Path: + return corpus.resolve_root(store, self.sc.package_root) + + def _instructions(self, cfg: DocumentCfg) -> str | None: + if not cfg.prompt_file: + return None + root = self.sc.package_root + if not root: + raise DocumentError("prompt_file requires ONTOGENY_PACKAGE_ROOT") + path = (Path(root) / cfg.prompt_file).resolve() + if not str(path).startswith(str(Path(root).resolve())): + raise DocumentError("prompt_file must stay inside the package root") + if not path.is_file(): + raise DocumentError(f"prompt_file not found: {cfg.prompt_file}") + return path.read_text(encoding="utf-8") + + # ------------------------------------------------------- default source + + async def ensure_default_source(self) -> dict[str, Any]: + """Create a generic document corpus when the package has none. + + This is an explicit admin action, not a silent page-load write: adding + an object type changes the ontology, so the caller must ask for it. + """ + existing = self.document_stores() + if existing: + return {"created": False, "store": existing[0].metadata.name} + root = self.sc.package_root + if not root: + raise DocumentError("ONTOGENY_PACKAGE_ROOT is not set; cannot create a default source") + package = Path(root) + stores_dir = package / "stores" + objects_dir = package / "objects" + stores_dir.mkdir(parents=True, exist_ok=True) + objects_dir.mkdir(parents=True, exist_ok=True) + store_name = "documents" + suffix = 2 + while (stores_dir / f"{store_name}.yaml").exists() or store_name in self.compiled.stores: + store_name = f"documents-{suffix}" + suffix += 1 + if (stores_dir / f"{store_name}.yaml").exists(): + raise DocumentError(f"store resource {store_name!r} already exists") + type_name = "document-note" + suffix = 2 + while (objects_dir / f"{type_name}.yaml").exists() or type_name in self.compiled.objects: + type_name = f"document-note-{suffix}" + suffix += 1 + store_yaml = f"""apiVersion: ontogeny/v1 +kind: Store +metadata: + name: {store_name} + display: Documents + description: Default document corpus for document intelligence +spec: + type: document + connection: data/docs + documents: + include: + - "**/*.md" + - "**/*.txt" + - "**/*.pdf" + - "**/*.docx" + split: paragraph + language: en +""" + type_yaml = f"""apiVersion: ontogeny/v1 +kind: ObjectType +metadata: + name: {type_name} + display: Document note + description: A fact extracted from a document +spec: + primaryKey: [note_id] + properties: + note_id: + type: string + display: Note ID + required: true + owner: document + title: + type: string + display: Title + required: true + owner: document + summary: + type: string + display: Summary + owner: document + body: + type: string + display: Body + owner: document + backing: + store: {store_name} + mode: materialized + source: + table: document-notes + sync: + strategy: extract + extract: + key: [title] + match: [title] + create: auto + update: fill-empty +""" + (stores_dir / f"{store_name}.yaml").write_text(store_yaml, encoding="utf-8") + (objects_dir / f"{type_name}.yaml").write_text(type_yaml, encoding="utf-8") + (package / "data" / "docs").mkdir(parents=True, exist_ok=True) + await self.sc.publish(package) + return {"created": True, "store": store_name, "object_type": type_name} + + # ---------------------------------------------------------------- corpus + + async def _sync_documents( + self, session, store, cfg, scanned: list[corpus.ScannedDocument] + ) -> tuple[list[DocumentRow], list[str]]: + rows = ( + (await session.execute(select(DocumentRow).where(DocumentRow.store == store.metadata.name))).scalars().all() + ) + by_path = {str(r.path): r for r in rows if r.status == "active"} + now = utcnow() + active: list[DocumentRow] = [] + changed: list[DocumentRow] = [] + seen: set[str] = set() + for item in scanned: + key = str(item.path) + seen.add(key) + row = by_path.get(key) + if row is not None and row.id == item.doc_id: + row.last_seen_at = now + row.status = "active" + active.append(row) + changed.append(row) + continue + if row is not None and row.id != item.doc_id: + # same path, new content: the old document is removed, the new + # one is a different id (content-addressed) + row.status = "removed" + changed.append(row) + new = DocumentRow( + id=item.doc_id, + store=store.metadata.name, + selector=None, + path=key, + media_type=item.media_type, + size=item.size, + status="active", + first_seen_at=now, + last_seen_at=now, + ) + session.add(new) + active.append(new) + changed.append(new) + removed: list[str] = [] + for row in rows: + if row.status == "active" and str(row.path) not in seen: + row.status = "removed" + removed.append(row.id) + changed.append(row) + return active, removed + + # ------------------------------------------------------------- extract + + def _options(self, cfg: DocumentCfg) -> ExtractOptions: + return ExtractOptions( + split=cfg.split, + language=cfg.language, + instructions=self._instructions(cfg), + max_chunks_per_doc=cfg.max_chunks_per_doc, + ) + + async def _packets_for( + self, session, doc_id: str, schema_hash: str, prompt_hash: str, model: str + ) -> list[ExtractPacketRow]: + rows = ( + ( + await session.execute( + select(ExtractPacketRow) + .where( + ExtractPacketRow.doc_id == doc_id, + ExtractPacketRow.schema_hash == schema_hash, + ExtractPacketRow.prompt_hash == prompt_hash, + ExtractPacketRow.model == model, + ) + .order_by(ExtractPacketRow.chunk_index) + ) + ) + .scalars() + .all() + ) + return list(rows) + + async def _extract_doc( + self, session, run, doc: DocumentRow, schema, options, *, schema_hash: str, prompt_hash: str, model: str + ) -> int: + extractor = self._extractor() + if extractor is None: + raise DocumentError( + "no document extractor is configured; install/enable extensions/doc-extract-agent and set " + "ONTOGENY_DOC_AGENT_CMD or ONTOGENY_DOC_AGENT_URL" + ) + path = Path(doc.path) + text, content_b64 = corpus.read_for_agent(path) + results = await extractor.extract( + ExtractRequest( + schema=schema, + docs=[ + DocInput( + doc_id=doc.id, + path=doc.path, + text=text, + name=path.name, + media_type=doc.media_type, + content_b64=content_b64, + ) + ], + options=options, + ) + ) + # replace this run's packets for the doc (a rerun of the same doc) + for result in results: + session.add( + ExtractPacketRow( + run_id=run.id, + doc_id=doc.id, + doc_hash=doc.id, + chunk_index=result.index, + chunk_hash=_sha(result.text), + schema_hash=schema_hash, + prompt_hash=prompt_hash, + model=model, + text=result.text, + graph=_jsonable(result.graph), + issues=_jsonable(result.issues), + status=result.status, + ) + ) + doc.chunk_count = len(results) + doc.extracted_at = utcnow() + doc.error = None + return len(results) + + # ---------------------------------------------------------------- bind + + async def _bind(self, session, run, packets: list[tuple[ExtractPacketRow, dict]], schema): + binder = Binder(self.compiled, schema, self.sc.repo, session, run_id=run.id) + return await binder.bind(packets) + + # --------------------------------------------------------------- apply + + async def _write_object( + self, + session, + bound: BoundObject, + *, + run_id: int, + force: bool = False, + principal: str, + created_by: str | None = None, + action: str = "ingest.extract", + ) -> str: + obj = self.compiled.objects[bound.object_type] + cfg = obj.spec.backing.extract or ExtractCfg() + bound.props.setdefault(obj.spec.primaryKey[0], bound.object_id) + current = await self.sc.repo.get(session, bound.object_type, bound.object_id) + ts = utcnow() + evidence = bound.evidence + doc_id = bound.created_by or (next(iter(bound.docs)) if bound.docs else "") + chunk = next(iter(evidence.values())).chunk_index if evidence else 0 + # two documents disagreed about one field: the first value is kept and + # the disagreement is surfaced instead of letting chunk order decide + for prop, values in bound.alternatives.items(): + if not values: + continue + # point the review at the document that supplied the disagreeing + # value, not at the one that happened to be parsed first + alt = bound.alternative_evidence.get(prop) + payload_evidence = ( + {prop: alt.as_dict()} if alt is not None else {k: v.as_dict() for k, v in evidence.items()} + ) + await self._candidate( + session, + run_id, + alt.doc_id if alt is not None else doc_id, + alt.chunk_index if alt is not None else chunk, + "object", + "CONFLICT", + { + "object_type": bound.object_type, + "object_id": bound.object_id, + "field": prop, + "current": _jsonable(bound.props.get(prop)), + "proposed": _jsonable(values[0]), + "alternatives": _jsonable(values), + "props": _jsonable(bound.props), + "evidence": payload_evidence, + }, + object_type=bound.object_type, + object_id=bound.object_id, + ) + + if current is None: + if bound.missing_required: + await self._candidate( + session, + run_id, + doc_id, + chunk, + "object", + "MISSING_REQUIRED", + { + "object_type": bound.object_type, + "object_id": bound.object_id, + "props": _jsonable(bound.props), + "missing": bound.missing_required, + "evidence": {k: v.as_dict() for k, v in evidence.items()}, + }, + object_type=bound.object_type, + object_id=bound.object_id, + ) + return "candidate" + if not force and cfg.create != "auto": + await self._candidate( + session, + run_id, + doc_id, + chunk, + "object", + "NEW_OBJECT", + { + "object_type": bound.object_type, + "object_id": bound.object_id, + "props": _jsonable(bound.props), + "evidence": {k: v.as_dict() for k, v in evidence.items()}, + }, + object_type=bound.object_type, + object_id=bound.object_id, + ) + return "candidate" + outcome = await self.sc.repo.apply_extracted(session, bound.object_type, bound.props, ts) + created = True + else: + values: dict[str, Any] = {} + for prop, value in bound.props.items(): + if prop in self.compiled.objects[bound.object_type].spec.primaryKey: + continue + if not _valid(value): + continue + owner = self.compiled.owner(bound.object_type, prop) + # even a human acceptance cannot write a system-of-record + # field: sync owns it, and the review gate is not a second + # sync with better intentions + if owner == "source" and not cfg.include_source: + continue + existing = current.get(self.compiled.col(prop)) + if force: + values[prop] = value + continue + if cfg.update == "never": + continue + if cfg.update == "fill-empty" and _valid(existing): + if existing != value: + await self._candidate( + session, + run_id, + doc_id, + chunk, + "object", + "CONFLICT", + { + "object_type": bound.object_type, + "object_id": bound.object_id, + "field": prop, + "current": _jsonable(existing), + "proposed": _jsonable(value), + "props": _jsonable(bound.props), + "evidence": {k: v.as_dict() for k, v in evidence.items()}, + }, + object_type=bound.object_type, + object_id=bound.object_id, + ) + continue + values[prop] = value + if not values: + return "noop" + values[obj.spec.primaryKey[0]] = bound.object_id + outcome = await self.sc.repo.apply_extracted(session, bound.object_type, values, ts) + created = False + + if created and outcome == "inserted": + # durable origin: assertions may later be deleted with their + # document, but "the corpus created this row" must survive + origin = await session.get(ExtractOriginRow, (bound.object_type, bound.object_id)) + if origin is None: + session.add( + ExtractOriginRow( + object_type=bound.object_type, + object_id=bound.object_id, + doc_id=doc_id, + run_id=run_id, + ) + ) + if outcome in ("inserted", "updated"): + payload_props = bound.props if created else values + session.add( + OutboxRow( + object_type=bound.object_type, + object_id=bound.object_id, + op="upsert", + payload={"props": _jsonable(payload_props), "origin": "document", "run_id": run_id}, + ) + ) + session.add( + RevisionRow( + action=action, + object_type=bound.object_type, + object_id=bound.object_id, + principal=principal, + params={"run_id": run_id, "doc_id": doc_id}, + before=_jsonable(current) if current else None, + after=_jsonable(bound.props), + outcome="executed", + message=f"extracted from {doc_id[:12]}" if doc_id else "extracted", + ) + ) + written = bound.props if created else values + for prop, value in written.items(): + docs = bound.sources.get(prop) or bound.docs or {doc_id} + for source_doc in docs: + ev = evidence.get(prop) + await session.execute( + delete(ExtractAssertionRow).where( + ExtractAssertionRow.doc_id == source_doc, + ExtractAssertionRow.kind == "object", + ExtractAssertionRow.object_type == bound.object_type, + ExtractAssertionRow.object_id == bound.object_id, + ExtractAssertionRow.prop == prop, + ) + ) + session.add( + ExtractAssertionRow( + run_id=run_id, + doc_id=source_doc, + chunk_index=ev.chunk_index if ev else 0, + kind="object", + object_type=bound.object_type, + object_id=bound.object_id, + prop=prop, + value=_jsonable(value), + evidence=ev.as_dict() if ev else {}, + created_object=bool(created and (created_by in (None, source_doc))), + ) + ) + return outcome + + async def _candidate( + self, + session, + run_id: int, + doc_id: str, + chunk: int, + kind: str, + reason: str, + payload: dict[str, Any], + *, + object_type: str | None = None, + link: str | None = None, + object_id: str | None = None, + ) -> None: + material = "|".join( + str(x) + for x in ( + kind, + reason, + object_type, + link, + object_id, + doc_id, + payload.get("field"), + payload.get("proposed"), + payload.get("src_id"), + payload.get("dst_id"), + ) + ) + signature = _sha(material) + existing = ( + ( + await session.execute( + select(ExtractCandidateRow) + .where( + ExtractCandidateRow.signature == signature, + ExtractCandidateRow.status == "pending", Review Comment: Important: the signature lookup matches only `pending` rows, so a rejected candidate is recreated as a new pending row on the next ingest of the unchanged corpus, and reviewers must reject it again on every run. Match `rejected` rows with the same signature too and skip recreating them. ########## ontogeny/server/ontogeny/documents/binder.py: ########## @@ -0,0 +1,518 @@ +# 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. + +"""Deterministic binder: extractor subgraph -> ontology objects and edges. + +No model here, and no I/O beyond lookups. Everything the extractor produced is +treated as *evidence*: values are coerced to the ontology type, identities are +resolved against live rows, and anything ambiguous becomes a review candidate +instead of a silent merge. +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import json +import re +from dataclasses import dataclass, field +from typing import Any + +from ..core.models import ExtractCfg, ExtractedJoin +from ..core.types import PropertyType +from ..errors import TypeMismatchError +from .schema import SchemaBundle + +_WS = re.compile(r"\s+") + + +@dataclass +class Evidence: + doc_id: str + chunk_index: int + quote: str + + def as_dict(self) -> dict[str, Any]: + return {"doc_id": self.doc_id, "chunk_index": self.chunk_index, "quote": self.quote} + + +@dataclass +class BoundObject: + object_type: str + object_id: str + props: dict[str, Any] + evidence: dict[str, Evidence] + created: bool + alternatives: dict[str, list[Any]] = field(default_factory=dict) + #: the document/chunk that supplied the disagreeing value + alternative_evidence: dict[str, Evidence] = field(default_factory=dict) + missing_required: list[str] = field(default_factory=list) + identity_source: str = "pk" + #: property -> documents that supplied it (provenance / retraction) + sources: dict[str, set[str]] = field(default_factory=dict) + docs: set[str] = field(default_factory=set) + created_by: str | None = None + + def payload(self) -> dict[str, Any]: + return { + "object_type": self.object_type, + "object_id": self.object_id, + "props": _jsonable(self.props), + "evidence": {k: v.as_dict() for k, v in self.evidence.items()}, + "created": self.created, + "missing_required": self.missing_required, + "identity_source": self.identity_source, + } + + +@dataclass +class BoundLink: + link: str + src_id: str + dst_id: str + evidence: Evidence + out_label: str + in_label: str + #: one evidence entry per asserting document, so retraction is per-doc + sources: dict[str, Evidence] = field(default_factory=dict) + + @property + def key(self) -> tuple[str, str, str]: + return (self.link, self.src_id, self.dst_id) + + def payload(self) -> dict[str, Any]: + return { + "link": self.link, + "src_id": self.src_id, + "dst_id": self.dst_id, + "source_label": self.out_label, + "target_label": self.in_label, + "evidence": self.evidence.as_dict(), + } + + +@dataclass +class Candidate: + kind: str # object | link + reason: str + payload: dict[str, Any] + doc_id: str + chunk_index: int + object_type: str | None = None + link: str | None = None + object_id: str | None = None + + +@dataclass +class Reject: + object_type: str + code: str + reason: str + source_pk: str | None + row: dict[str, Any] + + +@dataclass +class BindOutcome: + objects: dict[tuple[str, str], BoundObject] = field(default_factory=dict) + links: dict[tuple[str, str, str], BoundLink] = field(default_factory=dict) + candidates: list[Candidate] = field(default_factory=list) + rejected: list[Reject] = field(default_factory=list) + gaps: list[str] = field(default_factory=list) + #: (object_type, id) of DB rows the corpus pointed at (never created) + references: set[tuple[str, str]] = field(default_factory=set) + + +# --------------------------------------------------------------- coercion + + +def _canon(value: Any) -> Any: + if isinstance(value, _dt.datetime): + return value.isoformat() + if isinstance(value, _dt.date): + return value.isoformat() + if isinstance(value, dict): + return {str(k): _canon(v) for k, v in value.items()} + if isinstance(value, (list, tuple)): + return [_canon(v) for v in value] + return value + + +def _jsonable(value: Any) -> Any: + return _canon(value) + + +def _normalize_scalar(value: Any) -> Any: + if isinstance(value, str): + v = value.strip() + # currency/thousands separators are common in documents and never + # meaningful in a numeric ontology property + if re.fullmatch(r"[-+]?[$€£¥]?\s*\d[\d,]*\.?\d*%?", v): + v = v.replace(",", "").replace(" ", "") + for ch in "$€£¥": + v = v.replace(ch, "") + if v.endswith("%"): + try: + return float(v[:-1]) / 100.0 + except ValueError: + return v + return v + return v + return value + + +def coerce_value(ptype: PropertyType, value: Any, *, prop: str) -> Any: + """Coerce with the forgiving normalizations real documents need. + + The platform's own coercion is strict (an enum is an exact match); a model + reading "High" for `enum[LOW, HIGH]` is not evidence of a schema violation, + so case/whitespace are normalized *before* validation, never after. + """ + v = _normalize_scalar(value) Review Comment: Important: `_normalize_scalar` runs before the type check, so string properties are rewritten too: `"1,2,3"` is stored as `"123"`, `"15%"` as `"0.15"` and `"$1,200"` as `"1200"`. Apply the currency, separator and percent normalization only for numeric kinds. ########## ontogeny/server/ontogeny/documents/tasks.py: ########## @@ -0,0 +1,879 @@ +# 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. + +"""Task-oriented document intelligence. + +The legacy ingest path writes as soon as extraction succeeds. This module is +the gated path: a task (one document + one scope + one skill snapshot) runs an +agent attempt into staging, a human can edit the reviewed graph, and only an +explicit confirm binds and writes object/link rows + outbox events. +""" + +from __future__ import annotations + +import datetime as _dt +import hashlib +import json +from pathlib import Path +from types import SimpleNamespace +from typing import Any + +from sqlalchemy import func, select + +from ..errors import NotFoundError, OOError +from . import corpus, skills +from .binder import Binder +from .models import ( + DocumentAttemptRow, + DocumentEditRow, + DocumentRow, + DocumentTaskRow, + ExtractCandidateRow, + ExtractionSkillRow, +) +from .protocol import DocInput, ExtractOptions, ExtractRequest +from .schema import compile_extraction_schema + +BLOCKING = { + "NO_IDENTITY", + "AMBIGUOUS_MATCH", + "UNRESOLVED_REFERENCE", + "AMBIGUOUS_REFERENCE", + "UNRESOLVED_ENDPOINT", + "WRONG_DIRECTION", + "MISSING_REQUIRED", + "TYPE_MISMATCH", +} + + +class TaskError(OOError): + code = "DOCUMENT_TASK_ERROR" + http_status = 422 + + +def _sha(value: Any) -> str: + return hashlib.sha256( + json.dumps(value, sort_keys=True, ensure_ascii=False, default=str).encode("utf-8") + ).hexdigest() + + +def _json(value: Any) -> Any: + if isinstance(value, _dt.datetime | _dt.date): + return value.isoformat() + if isinstance(value, dict): + return {k: _json(v) for k, v in value.items()} + if isinstance(value, list | tuple): + return [_json(v) for v in value] + return value + + +class TaskService: + def __init__(self, docs) -> None: + self.docs = docs + self.sc = docs.sc + + # ------------------------------------------------------------- helpers + + async def _stored_skills_async(self) -> list[dict[str, Any]]: + async with self.sc.sessionmaker() as session: + rows = ( + (await session.execute(select(ExtractionSkillRow).where(ExtractionSkillRow.status == "active"))) + .scalars() + .all() + ) + return [ + { + "id": r.name, + "name": r.name, + "display": r.display or r.name, + "version": r.version, + "description": r.description, + "body": r.body, + "examples": r.examples or [], + } + for r in rows + ] + + def _targets( + self, store_name: str, scope: dict[str, Any] | None + ) -> tuple[list[str], dict[str, list[str]] | None, list[str] | None]: + defaults = self.docs.object_types(store_name) + if not scope: + return defaults, None, None + selected = scope.get("objects") + if not isinstance(selected, dict) or not selected: + raise TaskError("scope.objects must be a non-empty object map") + targets = sorted(selected) + unknown = [t for t in targets if t not in defaults] + if unknown: + raise TaskError(f"scope selects types not backed by store {store_name!r}: {unknown}") + properties = {name: list(vals) for name, vals in selected.items()} + links = scope.get("links") + if links is not None and not isinstance(links, list): + raise TaskError("scope.links must be an array") + return targets, properties, links + + def _reference_types(self, targets: list[str], links: list[str] | None) -> list[str]: + selected = set(links) if links is not None else None + out: list[str] = [] + for name, lnk in self.docs.compiled.links.items(): + if selected is not None and name not in selected: + continue + from ..core.models import ExtractedJoin + + if not isinstance(lnk.spec.join, ExtractedJoin): + continue + for endpoint in (lnk.spec.source, lnk.spec.target): + if endpoint in targets or endpoint in out: + continue + obj = self.docs.compiled.objects.get(endpoint) + b = obj.spec.backing if obj is not None else None + if b is not None and b.extract is not None and (b.extract.match or b.extract.key): + out.append(endpoint) + return out + + def _bundle(self, store_name: str, scope: dict[str, Any] | None): + targets, properties, links = self._targets(store_name, scope) + refs = self._reference_types(targets, links) + try: + return compile_extraction_schema( + self.docs.compiled, + targets, + reference_types=refs, + properties=properties, + links=links, + ) + except ValueError as exc: + raise TaskError(str(exc)) from exc + + def _doc(self, session, task: DocumentTaskRow) -> DocumentRow: + doc = session.get(DocumentRow, task.doc_id) + if doc is None: + raise NotFoundError(f"document {task.doc_id} not found") + return doc + + # -------------------------------------------------------------- skills + + async def list_skills(self) -> dict[str, Any]: + stored = await self._stored_skills_async() + return {"skills": [skills.GENERIC, *stored]} + + async def create_skill(self, body: dict[str, Any], principal: str) -> dict[str, Any]: + name = str(body.get("name") or "").strip() + if not name: + raise TaskError("skill name is required") + payload = skills.snapshot( + {"id": name, "addendum": body.get("body") or "", "examples": body.get("examples") or []} + ) + async with self.sc.sessionmaker() as session: + existing = ( + await session.execute(select(ExtractionSkillRow).where(ExtractionSkillRow.name == name)) + ).scalar_one_or_none() + if existing is not None: + raise OOError(f"skill {name!r} already exists", code="CONFLICT", http_status=409) + row = ExtractionSkillRow( + name=name, + version=1, + display=str(body.get("display") or name), + description=str(body.get("description") or ""), + body=str(body.get("body") or ""), + examples=body.get("examples") or [], + created_by=principal, + ) + session.add(row) + await session.commit() + return {"id": name, "name": name, "version": 1, "snapshot": payload} + + async def _snapshot(self, skill: dict[str, Any] | None) -> dict[str, Any]: + stored = await self._stored_skills_async() + try: + return skills.snapshot(skill, stored) + except skills.SkillError as exc: + raise TaskError(str(exc)) from exc + + # --------------------------------------------------------------- scope + + async def schema_options(self, store_name: str) -> dict[str, Any]: + self.docs.store(store_name) + targets = self.docs.object_types(store_name) + objects = [] + for name in targets: + obj = self.docs.compiled.objects[name] + cfg = obj.spec.backing.extract if obj.spec.backing is not None else None + identity = list(cfg.key or []) if cfg else [] + props = [] + for prop, d in obj.spec.properties.items(): + if d.derived is not None: + continue + if d.ptype().kind in ("struct", "vector"): + continue + props.append( + { + "name": prop, + "display": d.display, + "type": d.ptype().kind, + "required": bool(d.required), + "identity": prop in identity or prop == obj.spec.primaryKey[0], + "owner": d.owner, + } + ) + objects.append( + { + "name": name, + "display": obj.metadata.display or name, + "description": obj.metadata.description, + "identity": identity, + "properties": props, + } + ) + links = [] + from ..core.models import ExtractedJoin + + for name, lnk in sorted(self.docs.compiled.links.items()): + if not isinstance(lnk.spec.join, ExtractedJoin): + continue + if lnk.spec.target not in targets: + continue + source_ref = lnk.spec.source not in targets + links.append( + { + "name": name, + "display": lnk.metadata.display or name, + "source": lnk.spec.source, + "target": lnk.spec.target, + "source_is_reference": source_ref, + } + ) + return {"store": store_name, "objects": objects, "links": links} + + async def preview(self, store_name: str, scope: dict[str, Any] | None) -> dict[str, Any]: + bundle = self._bundle(store_name, scope) + return {"schema": bundle.schema, "schema_hash": bundle.hash} + + # --------------------------------------------------------------- tasks + + async def create_tasks( + self, + store_name: str, + files: list[dict[str, str]], + *, + doc_ids: list[str] | None = None, + scope: dict[str, Any] | None, + skill: dict[str, Any] | None, + principal: str, + ) -> dict[str, Any]: + self.docs.store(store_name) + if not files and not doc_ids: + raise TaskError("choose at least one file or existing document") + bundle = self._bundle(store_name, scope) + snapshot = await self._snapshot(skill) + written: list[str] = [] + if files: + uploaded = await self.docs._upload(store_name, files) + written = list(uploaded.get("written") or []) + await self.docs.scan_corpus(store_name) + root = self.docs._root(self.docs.store(store_name)).resolve() + paths = {(root / p).resolve() for p in written} + wanted_ids = set(doc_ids or []) + async with self.sc.sessionmaker() as session: + docs = (await session.execute(select(DocumentRow).where(DocumentRow.store == store_name))).scalars().all() + selected = [d for d in docs if d.id in wanted_ids or Path(d.path).resolve() in paths] + if not selected: + raise TaskError("no uploaded or existing document was found") + if wanted_ids: + missing = wanted_ids - {d.id for d in selected} + if missing: + raise TaskError(f"unknown document ids: {sorted(missing)}") + tasks = [] + for doc in selected: + existing = ( + ( + await session.execute( + select(DocumentTaskRow) + .where( + DocumentTaskRow.doc_id == doc.id, + DocumentTaskRow.status.in_(("draft", "review", "failed")), + ) + .order_by(DocumentTaskRow.id.desc()) + ) + ) + .scalars() + .first() + ) + if existing is None: + latest = ( + ( + await session.execute( + select(DocumentTaskRow) + .where(DocumentTaskRow.doc_id == doc.id) + .order_by(DocumentTaskRow.revision.desc(), DocumentTaskRow.id.desc()) + ) + ) + .scalars() + .first() + ) + existing = DocumentTaskRow( + store=store_name, + doc_id=doc.id, + status="draft", + revision=(latest.revision + 1) if latest is not None else 1, + supersedes_task_id=latest.id if latest is not None and latest.status == "confirmed" else None, + created_by=principal, + ) + session.add(existing) + existing.scope = scope + existing.schema_hash = bundle.hash + existing.skill = snapshot + existing.skill_hash = snapshot["hash"] + existing.error = None + existing.status = "draft" + await session.flush() + tasks.append(self._task_dict(existing, doc)) + await session.commit() + return {"tasks": tasks, "written": written, "schema_hash": bundle.hash} + + async def list_tasks(self, store_name: str | None = None, status: str | None = None) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + stmt = select(DocumentTaskRow).order_by(DocumentTaskRow.id.desc()).limit(500) + if store_name: + stmt = stmt.where(DocumentTaskRow.store == store_name) + if status: + stmt = stmt.where(DocumentTaskRow.status == status) + tasks = (await session.execute(stmt)).scalars().all() + docs = ( + { + d.id: d + for d in ( + await session.execute(select(DocumentRow).where(DocumentRow.id.in_([t.doc_id for t in tasks]))) + ).scalars() + } + if tasks + else {} + ) + return {"tasks": [self._task_dict(t, docs.get(t.doc_id)) for t in tasks]} + + def _task_dict(self, task: DocumentTaskRow, doc: DocumentRow | None = None) -> dict[str, Any]: + return { + "id": task.id, + "store": task.store, + "doc_id": task.doc_id, + "name": Path(doc.path).name if doc is not None else task.doc_id, + "path": doc.path if doc is not None else None, + "status": task.status, + "revision": task.revision, + "supersedes_task_id": task.supersedes_task_id, + "scope": task.scope, + "schema_hash": task.schema_hash, + "skill": task.skill, + "skill_hash": task.skill_hash, + "current_attempt_id": task.current_attempt_id, + "confirmed_attempt_id": task.confirmed_attempt_id, + "created_by": task.created_by, + "created_at": task.created_at.isoformat() if task.created_at else None, + "confirmed_by": task.confirmed_by, + "confirmed_at": task.confirmed_at.isoformat() if task.confirmed_at else None, + "graph_sync_status": task.graph_sync_status, + "confirm_summary": task.confirm_summary, + "error": task.error, + } + + def _attempt_dict(self, attempt: DocumentAttemptRow, *, include_graph: bool = True) -> dict[str, Any]: + data = { + "id": attempt.id, + "task_id": attempt.task_id, + "attempt_no": attempt.attempt_no, + "status": attempt.status, + "agent_run_id": attempt.agent_run_id, + "provider": attempt.provider, + "model": attempt.model, + "schema_hash": attempt.schema_hash, + "prompt_hash": attempt.prompt_hash, + "chunks": attempt.chunks or [], + "issues": attempt.issues or [], + "validation": attempt.validation or {}, + "error": attempt.error, + "started_at": attempt.started_at.isoformat() if attempt.started_at else None, + "finished_at": attempt.finished_at.isoformat() if attempt.finished_at else None, + } + if include_graph: + data["graph"] = attempt.reviewed_graph or attempt.raw_graph or {"vertices": [], "edges": []} + data["raw_graph"] = attempt.raw_graph or {"vertices": [], "edges": []} + return data + + async def detail(self, task_id: int) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + task = await session.get(DocumentTaskRow, task_id) + if task is None: + raise NotFoundError(f"no task {task_id}") + doc = await session.get(DocumentRow, task.doc_id) + attempts = ( + ( + await session.execute( + select(DocumentAttemptRow) + .where(DocumentAttemptRow.task_id == task_id) + .order_by(DocumentAttemptRow.attempt_no) + ) + ) + .scalars() + .all() + ) + current = next((a for a in attempts if a.id == task.current_attempt_id), attempts[-1] if attempts else None) + data = self._task_dict(task, doc) + data["chunks"] = current.chunks if current is not None else [] + data["attempt"] = self._attempt_dict(current) if current is not None else None + data["attempts"] = [self._attempt_dict(a, include_graph=False) for a in attempts] + return data + + # ---------------------------------------------------------------- agent + + async def _agent_task(self, payload: dict[str, Any]) -> dict[str, Any]: + extractor = self.docs._extractor() + if extractor is None: + raise TaskError("no document extractor is configured") + doc = payload["document"] + request = ExtractRequest( + schema=payload["schema"], + docs=[ + DocInput( + doc_id=doc["doc_id"], + path=doc["path"], + text=doc.get("text") or "", + name=doc.get("name") or "", + media_type=doc.get("media_type"), + content_b64=doc.get("content_b64"), + ) + ], + options=ExtractOptions( + split=payload.get("split", "paragraph"), + language=payload.get("language", "en"), + max_chunks_per_doc=int(payload.get("max_chunks_per_doc", 200)), + instructions=payload.get("instructions"), + ), + ) + results = await extractor.extract(request) + return { + "chunks": [ + { + "doc_id": r.doc_id, + "index": r.index, + "text": r.text, + "graph": r.graph, + "issues": r.issues, + "status": r.status, + "reason": r.reason, + } + for r in results + ] + } + + async def _run_agent(self, payload: dict[str, Any], *, principal: str, task: str) -> dict[str, Any]: + # The extractor runs out-of-process over the doc-agent protocol; the + # agent broker's governed surface (sessions + call_tool) is for + # ontology tools, not extraction payloads. + return {"output": await self._agent_task(payload), "agent_run_id": None, "provider": "direct"} + + # ---------------------------------------------------------------- run + + async def run(self, task_id: int, principal: str) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + task = await session.get(DocumentTaskRow, task_id) + if task is None: + raise NotFoundError(f"no task {task_id}") + if task.status == "confirmed": + raise TaskError("confirmed task is immutable; rerun creates a new revision") + doc = await session.get(DocumentRow, task.doc_id) + if doc is None: + raise NotFoundError(f"document {task.doc_id} not found") + bundle = self._bundle(task.store, task.scope) + last = ( + await session.execute( + select(func.max(DocumentAttemptRow.attempt_no)).where(DocumentAttemptRow.task_id == task_id) + ) + ).scalar_one() + attempt = DocumentAttemptRow( + task_id=task_id, + attempt_no=int(last or 0) + 1, + status="running", + schema_hash=bundle.hash, + prompt_hash=task.skill_hash, + ) + session.add(attempt) + await session.flush() + task.current_attempt_id = attempt.id + task.status = "running" + task.schema_hash = bundle.hash + task.error = None + await session.commit() + task_id, attempt_id = task.id, attempt.id + doc_payload = { + "doc_id": doc.id, + "path": doc.path, + "name": Path(doc.path).name, + "media_type": doc.media_type, + "text": "", + "content_b64": None, + } + path = Path(doc.path) + text, content_b64 = corpus.read_for_agent(path) + doc_payload["text"] = text + doc_payload["content_b64"] = content_b64 + skill = task.skill or {} + cfg = self.docs._cfg(self.docs.store(task.store)) + payload = { + "schema": bundle.schema, + "document": doc_payload, + "split": cfg.split, + "language": skill.get("language") or cfg.language, + "max_chunks_per_doc": cfg.max_chunks_per_doc, + "instructions": skills.instructions(skill) if skill else self.docs._instructions(cfg), + } + try: + result = await self._run_agent(payload, principal=principal, task=f"task:{task_id}") + chunks = list((result.get("output") or {}).get("chunks") or []) + vertices: list[Any] = [] + edges: list[Any] = [] + seen_vertices: set[str] = set() + seen_edges: set[tuple[str, str, str]] = set() + for chunk in chunks: + graph = chunk.get("graph") if isinstance(chunk, dict) else None + if isinstance(graph, dict): + chunk_vertices = graph.get("vertices") or [] + chunk_edges = graph.get("edges") or [] + elif isinstance(chunk, dict) and isinstance(chunk.get("vertices"), list): + chunk_vertices = chunk.get("vertices") or [] + chunk_edges = chunk.get("edges") or [] + else: + continue + for vertex in chunk_vertices: + if not isinstance(vertex, dict): + vertices.append(vertex) + continue + # the same reference may be named by every table row; one + # graph vertex is enough, and duplicate ids crash G6 + key = f"{vertex.get('label')}/{vertex.get('id')}" + if key in seen_vertices: + continue + seen_vertices.add(key) + vertices.append(vertex) + for edge in chunk_edges: + if isinstance(edge, dict): + key = ( + str(edge.get("label")), + str(edge.get("outV")), + str(edge.get("inV")), + ) + if key in seen_edges: + continue + seen_edges.add(key) + edges.append(edge) + graph = {"vertices": vertices, "edges": edges} + validation = await self._validate(task_id, graph, bundle) + async with self.sc.sessionmaker() as session: + attempt = await session.get(DocumentAttemptRow, attempt_id) + task = await session.get(DocumentTaskRow, task_id) + attempt.status = "review" + attempt.agent_run_id = result.get("agent_run_id") + attempt.provider = result.get("provider") + attempt.raw_graph = graph + attempt.reviewed_graph = graph + attempt.chunks = chunks + attempt.validation = validation + attempt.finished_at = _dt.datetime.now(_dt.UTC) + task.status = "review" + await session.commit() + return await self.detail(task_id) + except Exception as exc: + async with self.sc.sessionmaker() as session: + attempt = await session.get(DocumentAttemptRow, attempt_id) + task = await session.get(DocumentTaskRow, task_id) + if attempt is not None: + attempt.status = "failed" + attempt.error = f"{type(exc).__name__}: {exc}"[:4000] + attempt.finished_at = _dt.datetime.now(_dt.UTC) + if task is not None: + task.status = "failed" + task.error = f"{type(exc).__name__}: {exc}"[:4000] + await session.commit() + raise + + # ----------------------------------------------------------- validation + + async def _validate(self, task_id: int, graph: dict[str, Any], bundle) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + task = await session.get(DocumentTaskRow, task_id) + if task is None: + raise NotFoundError(f"no task {task_id}") + doc = await session.get(DocumentRow, task.doc_id) + text = "" + async with self.sc.sessionmaker() as s: + attempt = await s.get(DocumentAttemptRow, task.current_attempt_id) if task.current_attempt_id else None + if attempt is not None: + text = "\n\n".join(str(c.get("text") or "") for c in (attempt.chunks or []) if isinstance(c, dict)) + if not text and doc is not None: + text, _ = corpus.read_for_agent(Path(doc.path)) + packet = SimpleNamespace(doc_id=task.doc_id, chunk_index=0, text=text) + binder = Binder(self.docs.compiled, bundle, self.docs.sc.repo, session, run_id=0) + outcome = await binder.bind([(packet, graph)]) + blockers = [] + warnings = [] + for cand in outcome.candidates: + item = { + "reason": cand.reason, + "kind": cand.kind, + "object_type": cand.object_type, + "link": cand.link, + "payload": _json(cand.payload), + } + (blockers if cand.reason in BLOCKING else warnings).append(item) + for reject in outcome.rejected: + blockers.append( + {"reason": reject.code, "kind": "object", "message": reject.reason, "payload": _json(reject.row)} + ) + for gap in outcome.gaps: + warnings.append({"reason": "SCHEMA_GAP", "message": gap}) + return { + "blockers": blockers, + "warnings": warnings, + "objects": len(outcome.objects), + "links": len(outcome.links), + "references": [list(x) for x in outcome.references], + "gaps": outcome.gaps[:50], + } + + # --------------------------------------------------------------- edit + + async def save_graph(self, task_id: int, graph: dict[str, Any], principal: str) -> dict[str, Any]: + if ( + not isinstance(graph, dict) + or not isinstance(graph.get("vertices"), list) + or not isinstance(graph.get("edges"), list) + ): + raise TaskError("graph must contain vertices[] and edges[]") + async with self.sc.sessionmaker() as session: + task = await session.get(DocumentTaskRow, task_id) + if task is None: + raise NotFoundError(f"no task {task_id}") + if task.status == "confirmed": + raise TaskError("confirmed task is immutable") + attempt = ( + await session.get(DocumentAttemptRow, task.current_attempt_id) if task.current_attempt_id else None + ) + if attempt is None: + raise TaskError("run the task before editing its graph") + before = attempt.reviewed_graph + attempt.reviewed_graph = graph + attempt.validation = await self._validate(task_id, graph, self._bundle(task.store, task.scope)) + session.add( + DocumentEditRow( + attempt_id=attempt.id, + actor=principal, + op="replace_graph", + path="graph", + before=before, + after=graph, + ) + ) + task.status = "review" + await session.commit() + return await self.detail(task_id) + + # ------------------------------------------------------------- confirm + + async def confirm(self, task_id: int, principal: str) -> dict[str, Any]: + async with self.sc.sessionmaker() as session: + task = await session.get(DocumentTaskRow, task_id) + if task is None: + raise NotFoundError(f"no task {task_id}") + if task.status == "confirmed": + return {"task": await self.detail(task_id), "already": True} + if task.current_attempt_id is None: + raise TaskError("task has no attempt to confirm") + attempt = await session.get(DocumentAttemptRow, task.current_attempt_id) + if attempt is None: + raise TaskError("current attempt is missing") + graph = attempt.reviewed_graph or attempt.raw_graph or {"vertices": [], "edges": []} + if not (graph.get("vertices") or graph.get("edges")): + raise TaskError("graph is empty; nothing to confirm") + bundle = self._bundle(task.store, task.scope) + validation = await self._validate(task_id, graph, bundle) + if validation["blockers"]: + raise TaskError( + "task has unresolved blockers; fix the graph before confirming", + details={"blockers": validation["blockers"]}, + ) + doc = await session.get(DocumentRow, task.doc_id) + text = "\n\n".join(str(c.get("text") or "") for c in (attempt.chunks or []) if isinstance(c, dict)) + if not text and doc is not None: + text, _ = corpus.read_for_agent(Path(doc.path)) + packet = SimpleNamespace(doc_id=task.doc_id, chunk_index=0, text=text) + binder = Binder(self.docs.compiled, bundle, self.docs.sc.repo, session, run_id=attempt.id) + outcome = await binder.bind([(packet, graph)]) + applied: set[tuple[str, str]] = set() + summary = {"inserted": 0, "updated": 0, "noop": 0, "links": 0, "candidates": 0} + for bound in outcome.objects.values(): + current = await self.docs.sc.repo.get(session, bound.object_type, bound.object_id) + if current is not None: + applied.add((bound.object_type, bound.object_id)) + result = await self.docs._write_object( + session, + bound, + run_id=attempt.id, Review Comment: Important: `confirm` stores `attempt.id` in `run_id` fields that otherwise hold `ExtractRunRow` ids, so `summary["candidates"]` also counts the pending candidates of ingest run N when attempt N confirms, and `list_candidates(store=...)` files task candidates under whichever store owns that run. Give task-path rows their own `ExtractRunRow` (or a separate attempt column) instead of reusing `attempt.id`. -- 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]
