orchestrata_ultimate.session.governance

Reference › orchestrata_ultimate › session › governance

SessionGovernance lets the Prolog rule base decide whether a session may open or close. Python observes facts about the session, such as whether work was logged or tests pass, and governance.pl holds the policy that combines them. The SessionController calls may_close from its checklist validation.

The Prolog bridge is created lazily on first use, so building a controller that never reaches a lifecycle boundary costs no Prolog startup; a bridge can also be injected for tests. _ready loads the governance rules through governance.prolog_client, and if that fails it logs a warning and returns nothing. In that case may_open and may_close return None ("not evaluable") and the caller's own checklist stands. may_open(session_id, has_goals) asserts or retracts the session_has_goals fact and asks can_open_session/1. may_close does the same for session_checklist_complete, session_tests_passing and session_artifacts_valid and asks can_close_session/1. Facts are re-asserted idempotently (retract all, then assert) so a session validated twice never keeps a stale answer. The session id is checked against a safe pattern before it is quoted as a Prolog atom, and an unsafe id raises ValueError. See the governance package for the rule base.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionGovernance

Line 65

Asks governance.pl whether a session may open or close.

Source

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

"""Session lifecycle governance — the rule base decides open/close.

``governance.pl`` has modelled the session lifecycle since it was written::

    can_open_session(SessionId) :-
        session_has_goals(SessionId),
        \\+ session_blacklisted(SessionId),
        session_requirements_met(SessionId, basic).

    can_close_session(SessionId) :-
        session_checklist_complete(SessionId),
        session_tests_passing(SessionId),
        session_artifacts_valid(SessionId),
        \\+ session_has_blocking_issues(SessionId).

None of it ever ran. Measured 2026-08-21: ``can_open_session/1`` and
``can_close_session/1`` were queried by nothing, and the facts they consume
were asserted by nothing — ``PrologBridge.start_session``/``end_session``,
the only code that asserts them, has **zero production callers**. The chain
was dead at both ends while ``SessionController._validate_checklist``
re-implemented the same policy as five hardcoded Python conditions, which is
the "no hardcoded rules" half of Constitutional Article II.

This module is the seam. Python contributes OBSERVATIONS about session state
— it is the only thing that can see whether a work log is empty — and the
rule base contributes the POLICY that combines them. That split is why the
switch is behaviour-preserving rather than hopeful:

    session_checklist_complete  <-  work logged AND choices AND handoff set
    session_tests_passing       <-  the test command passed
    session_artifacts_valid     <-  artifacts recorded
    session_has_blocking_issues <-  never asserted here; it is a rule over
                                    session_issue/2, which carries no facts,
                                    so it is false by default

Substituting those into ``can_close_session/1`` yields exactly the five
conditions ``_validate_checklist`` already enforced. The gain is not a
different answer today — it is that changing the policy is now a rule edit
instead of a code edit, and that ``session_has_blocking_issues/1`` becomes a
real extension point: assert a critical ``session_issue/2`` and the session
cannot close, with no Python change at all.

Availability posture: if the rule base cannot be consulted, every method
returns ``None`` — "could not evaluate" — and the caller keeps its own
answer. Denying every session close because a consult failed would trade a
governance gap for an outage, and this is the developer's own tooling.
"""

from __future__ import annotations

import logging
import re

from orchestrata_ultimate.core.infra_config import PROLOG_RULES_DIR
from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

_logger = logging.getLogger(__name__)

#: Session ids reach Prolog as atoms. Anything outside this shape is refused
#: rather than quoted-and-hoped: a session id is generated, never
#: user-authored, so a surprising one means something upstream is wrong.
_SAFE_ID = re.compile(r"^[A-Za-z0-9_.:-]{1,128}$")


class SessionGovernance:
    """Asks ``governance.pl`` whether a session may open or close."""

    def __init__(self, bridge: PrologBridge | None = None) -> None:
        """
        Args:
            bridge: Pre-built bridge (tests inject a scoped one). When None,
                the engine is built and the rules consulted LAZILY on first
                use, so constructing a controller that never reaches a
                lifecycle boundary costs no Prolog startup.
        """
        self._bridge = bridge
        self._loaded = bridge is not None

    def _ready(self) -> PrologBridge | None:
        if self._bridge is not None and self._loaded:
            return self._bridge
        try:
            from orchestrata_ultimate.governance.prolog_client import (
                load_governance_rules,
            )

            if self._bridge is None:
                self._bridge = PrologBridge()
            load_governance_rules(self._bridge, PROLOG_RULES_DIR)
            self._loaded = True
            return self._bridge
        except Exception as exc:
            _logger.warning(
                "session governance rule base unavailable (%s) — the caller's "
                "own checklist stands for this call",
                exc,
            )
            return None

    @staticmethod
    def _atom(session_id: str) -> str:
        if not _SAFE_ID.match(session_id):
            raise ValueError(f"unsafe session id for a Prolog atom: {session_id!r}")
        return f"'{session_id}'"

    def _assert(self, bridge: PrologBridge, fact: str, atom: str) -> None:
        """Re-assert a session fact idempotently.

        ``retractall`` first: a session that is validated twice (a failed
        ``end`` followed by a fixed one) must not leave the FIRST run's
        stale answer behind for the rule to read.
        """
        bridge.query(f"retractall(governance:{fact}({atom}))")
        bridge.query(f"assertz(governance:{fact}({atom}))")

    def _retract(self, bridge: PrologBridge, fact: str, atom: str) -> None:
        bridge.query(f"retractall(governance:{fact}({atom}))")

    def may_open(self, session_id: str, *, has_goals: bool) -> bool | None:
        """Whether ``can_open_session/1`` holds. None if not evaluable.

        ``has_goals`` comes from the session's purpose: the CLI requires
        ``--purpose`` on every ``begin``, so a legitimate session always has
        one and this cannot refuse work that used to succeed.
        """
        bridge = self._ready()
        if bridge is None:
            return None
        atom = self._atom(session_id)
        if has_goals:
            self._assert(bridge, "session_has_goals", atom)
        else:
            self._retract(bridge, "session_has_goals", atom)
        return bridge.query_bool(f"governance:can_open_session({atom})")

    def may_close(
        self,
        session_id: str,
        *,
        checklist_complete: bool,
        tests_passing: bool,
        artifacts_valid: bool,
    ) -> bool | None:
        """Whether ``can_close_session/1`` holds. None if not evaluable."""
        bridge = self._ready()
        if bridge is None:
            return None
        atom = self._atom(session_id)
        for fact, holds in (
            ("session_checklist_complete", checklist_complete),
            ("session_tests_passing", tests_passing),
            ("session_artifacts_valid", artifacts_valid),
        ):
            if holds:
                self._assert(bridge, fact, atom)
            else:
                self._retract(bridge, fact, atom)
        return bridge.query_bool(f"governance:can_close_session({atom})")