orchestrata_ultimate.session.kg_sync

Reference › orchestrata_ultimate › session › kg_sync › init

KnowledgeGraphSyncer writes the project's state into the knowledge graph in stages. It is the class that combines the stage mixins and the connection base: _base (connection and Cypher helpers), _stage_session, _stage_folders, _stage_code, _stage_specs_traces and _stage_docs. The CLI sync command in cli/kg drives it, and kg_context reads the result.

sync_all(session_seq) runs the stages in order: project, folders, code, specs, traces, docs and file traceability. When a session sequence number is given it also inserts the session stage after the project and appends the session-files stage at the end. Each stage returns a SyncStats; an exception in one stage is caught and recorded as an error in that stage's stats, so the later stages still run. The method returns a dictionary of stage name to SyncStats.

At a glance

Class diagram

Dependency diagram

API

Classes

KnowledgeGraphSyncer

Bases: _FoldersMixin, _CodeMixin, _SpecsTracesMixin, _SessionMixin, _DocsMixin, _AstHelpersMixin, _SyncerBase · Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-043']) · Line 48

Syncs project state to graph knowledge graph.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/session/kg_sync/__init__.py

# mypy: ignore-errors
"""Knowledge Graph Syncer — 8-stage sync pipeline.

Syncs full project state to the graph: project, sessions, folders, code,
specs, traces, session-file links, and documentation coverage.

Uses sync graph driver (not async) since CLI and hooks are synchronous.
All operations use MERGE for idempotency.

Spec: 006 - Session Management

Decomposed into responsibility mixins: stage mixins coordinate via the
shared ``self._repo`` / ``self._run`` owned by ``_SyncerBase`` (composed
LAST in the MRO).
"""

from orchestrata_ultimate.core.tracing.traceability import traces
from orchestrata_ultimate.session.kg_sync._ast_helpers import _AstHelpersMixin
from orchestrata_ultimate.session.kg_sync._base import _SyncerBase
from orchestrata_ultimate.session.kg_sync._common import (
    BATCH_SIZE,
    CODE_EXTENSIONS,
    IGNORE_DIRS,
    SYNC_EXTENSIONS,
    SyncStats,
)
from orchestrata_ultimate.session.kg_sync._stage_code import _CodeMixin
from orchestrata_ultimate.session.kg_sync._stage_docs import _DocsMixin
from orchestrata_ultimate.session.kg_sync._stage_folders import _FoldersMixin
from orchestrata_ultimate.session.kg_sync._stage_session import _SessionMixin
from orchestrata_ultimate.session.kg_sync._stage_specs_traces import _SpecsTracesMixin

__all__ = [
    "BATCH_SIZE",
    "CODE_EXTENSIONS",
    "IGNORE_DIRS",
    "SYNC_EXTENSIONS",
    "KnowledgeGraphSyncer",
    "SyncStats",
]


@traces(
    spec="008",
    story="US-001",
    reqs=["REQ-008-043"],
)
class KnowledgeGraphSyncer(
    _FoldersMixin,
    _CodeMixin,
    _SpecsTracesMixin,
    _SessionMixin,
    _DocsMixin,
    _AstHelpersMixin,
    _SyncerBase,
):
    """Syncs project state to graph knowledge graph.

    8-stage pipeline:
        0. project   — Root (:Project) node
        1. session   — (:Session) + link to Project
        2. folders   — (:Directory) tree + (:File) nodes
        3. code      — (:Class), (:Function) via AST + DEFINES
        4. specs     — (:Spec), (:Requirement), (:RequirementItem)
        5. traces    — TRACES_TO from code to requirements
        6. session_files — MODIFIED/PRODUCED from session to files
        7. docs      — has_docstring boolean on all Class/Function nodes
    """

    # ── Full sync ───────────────────────────────────────────

    def sync_all(self, session_seq: int | None = None) -> dict[str, SyncStats]:
        """Run all 8 sync stages in order.

        Args:
            session_seq: If given, sync this session and its file links.

        Returns:
            Dict mapping stage name to SyncStats.
        """
        results: dict[str, SyncStats] = {}
        stages = [
            ("project", self.sync_project),
            ("folders", self.sync_folders),
            ("code", lambda: self.sync_code()),
            ("specs", self.sync_specs),
            ("traces", self.sync_traces),
            ("docs", self.sync_docs),
            ("file_traceability", self._update_file_traceability),
        ]

        if session_seq is not None:
            stages.insert(1, ("session", lambda: self.sync_session(session_seq)))
            stages.append(
                ("session_files", lambda: self.sync_session_files(session_seq))
            )

        for name, fn in stages:
            try:
                stats = fn()
                results[name] = stats
            except Exception as e:
                results[name] = SyncStats(stage=name, errors=[str(e)])

        return results