mirror_observer.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/mirror_observer.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/unit/test_257_dispatcher_seams.py"
"""
Registry-side listener for the dispatch layer's plan-state events.
 
Purpose
-------
Plan 257 P1 (A7, cut 1). This module owns the legacy-compat mirroring
that used to live inside ``oc_chat/scripts/plan_status.py``: opening the
orchestration registry, taking the legacy-compat write lock, and
replaying a worker's ``.PLAN_STATUS.md`` lifecycle into the ``plans``
table. The code was moved here verbatim, not rewritten -- the point of
the move is that the dispatch layer stops naming the registry, not that
the mirroring behaves differently.
 
Responsibilities
----------------
- :class:`RegistryMirrorObserver` -- implements
  :class:`oc_chat.core.state_observer.StateObserver` against the live
  orchestration registry.
- :func:`compat_context` -- the fail-soft registry handle the mirror
  runs on: ``None`` when there is no registry to write to.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - lifecycle events (mirror attempts, drift reports)
    L3 - semantic details (skipped mirrors, absent registry)
    L4 - deep tracing (module load)
 
Contracts
---------
- Behaviour is byte-identical to the pre-257 ``plan_status`` mirror: the
  same ``REGISTRY DRIFT - <operation> failed: <Type>: <msg>. Run `oct
  agent migrate verify`.`` line on stderr, the same return code ``2``,
  and the same silent ``0`` when no registry file exists.
- A missing registry is not an error. A registry that exists but has no
  active operator slot IS -- that refusal is deliberately re-raised out
  of :func:`compat_context` and reported as drift.
- No exception escapes a callback. Every failure becomes the drift line
  plus return code ``2``.
- The registry handle is injectable so a witness can drive every branch
  without a SQLite file.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry (lazy; Registry facade opened by compat_context)
- oc_chat.registry.backends.sqlite (lazy; SQLiteBackend and the active operator slot in compat_context)
- oc_chat.registry.discovery (lazy; discover_registry_location in compat_context)
- oc_chat.registry.domain (lazy; plan_create, plan_assign and plan_set_status writers for the mirrors)
- oc_chat.scripts.plan_status (lazy; _plan_number for the plan_created mirror)
"""
 
from __future__ import annotations
 
import sys
from typing import Callable
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "mirror_observer loaded", 4)
 
 
#: What a mirror needs to write: an open registry, its backend, and the
#: actor slot the write is attributed to. ``None`` means "no registry
#: here", which is a silent skip rather than a failure.
CompatContext = Callable[[], "tuple[object, object, str] | None"]
 
 
def compat_context() -> tuple[object, object, str] | None:
    """Open the discovered orchestration registry, or return ``None``.
 
    Moved verbatim from ``plan_status._compat_context`` (plan 257 P1).
    The broad ``except`` is deliberate and pre-existing: a worker must be
    able to record its own status in a checkout that has no registry at
    all. The one refusal that is NOT swallowed is a registry that exists
    but has no active operator slot, because writing unattributed rows
    would be worse than refusing.
    """
    func = "compat_context"
    try:
        from oc_chat.registry import Registry
        from oc_chat.registry.backends.sqlite import SQLiteBackend
        from oc_chat.registry.discovery import discover_registry_location
 
        location = discover_registry_location()
        if not location.registry_path.is_file():
            return None
        backend = SQLiteBackend(location.registry_path, check_capabilities=False)
        actor_slot = backend.active_operator_slot()
        if actor_slot is None:
            raise RuntimeError("no active operator slot found")
        return Registry(backend), backend, actor_slot
    except Exception as exc:
        if "no active operator slot" in str(exc):
            raise
        _dbg("OC_CHAT.MAIN", func, f"no registry to mirror into: {exc}", 3)
        return None
 
 
def registry_plan_exists(registry: object, plan_id: str) -> bool:
    """True when *plan_id* already has a row in the registry."""
    return registry_plan_status(registry, plan_id) is not None
 
 
def registry_plan_status(registry: object, plan_id: str) -> str | None:
    """The registry's current status for *plan_id*, or ``None``."""
    row = registry_plan_row(registry, plan_id)
    return None if row is None else str(row["status"])
 
 
def registry_plan_row(registry: object, plan_id: str) -> object | None:
    """The registry's ``plans`` row for *plan_id*, or ``None``."""
    connect = getattr(getattr(registry, "backend", None), "connect", None)
    if connect is None:
        return None
    with connect() as conn:
        row = conn.execute(
            # OCT-LINT: disable=no_hardcoded_secrets reason="Status enum values, model IDs, and CLI choice strings are not secrets"
            "SELECT status, assigned_slot, queue_position, branch, worktree "
            "FROM plans WHERE plan_id = ?",
            (plan_id,),
        ).fetchone()
    return row
 
 
class RegistryMirrorObserver:
    """Mirrors dispatch-layer plan events into the orchestration registry.
 
    The status tool registers one of these by default, so a worker that
    changes nothing sees exactly the pre-257 behaviour. Passing
    ``context_factory`` swaps the registry handle, which is how the
    witnesses drive the drift and skip branches without a SQLite file.
    """
 
    def __init__(self, *, context_factory: CompatContext | None = None) -> None:
        self._context_factory: CompatContext = context_factory or compat_context
 
    def plan_created(
        self,
        *,
        plan_id: str,
        title: str,
        plan_type: str | None,
        body_path: str,
        body_sha256: str,
        body_observed_at: str | None,
        model: str,
        thinking: str,
        session: str,
        branch: str,
        worktree: str,
        target_status: str,
    ) -> int:
        """Create the plan row, re-anchor its assignment, set its status."""
        func = "plan_created"
        _dbg("OC_CHAT.MAIN", func, f"mirroring init for {plan_id}", 2)
 
        def _mirror(registry: object, actor_slot: str) -> None:
            from oc_chat.registry.domain import plan_assign, plan_create, plan_set_status
 
            # Derived here rather than by the caller so a plan id with no
            # numeric prefix still reports as drift and returns 2, exactly as
            # it did when this whole block lived inside the status tool. The
            # registry -> dispatch direction is the permitted one and is
            # already precedented by registry/migrate.py.
            from oc_chat.scripts.plan_status import _plan_number
 
            plan_create(
                registry,
                plan_id=plan_id,
                plan_number=_plan_number(plan_id),
                title=title,
                plan_type=plan_type,
                body_path=body_path,
                body_sha256=body_sha256,
                body_observed_at=body_observed_at,
                model_override=model,
                thinking_override=thinking,
                session_override=session,
                actor_slot=actor_slot,
                actor_type="human_operator",
                command="plan_status init",
            )
            plan_row = registry_plan_row(registry, plan_id)
            if (
                plan_row is not None
                and plan_row["assigned_slot"] is not None
                and plan_row["queue_position"] is not None
                and (
                    str(plan_row["branch"] or "") != branch
                    or str(plan_row["worktree"] or "") != worktree
                )
            ):
                plan_assign(
                    registry,
                    plan_id=plan_id,
                    from_slot=str(plan_row["assigned_slot"]),
                    to_slot=str(plan_row["assigned_slot"]),
                    branch=branch,
                    worktree=worktree,
                    queue_position=int(plan_row["queue_position"]),
                    actor_slot=actor_slot,
                    actor_type="human_operator",
                    command="plan_status init",
                )
            from_status = registry_plan_status(registry, plan_id) or "draft"
            if target_status != from_status:
                plan_set_status(
                    registry,
                    plan_id=plan_id,
                    from_status=from_status,
                    to_status=target_status,
                    source="worker_self",
                    note=(
                        "legacy plan_status init"
                        if target_status in {"partial", "failed"}
                        else None
                    ),
                    actor_slot=actor_slot,
                    actor_type="human_operator",
                    command="plan_status init",
                )
 
        return self._run("plan_status.init", _mirror)
 
    def status_changed(
        self,
        *,
        plan_id: str,
        from_status: str,
        to_status: str,
        note: str | None,
    ) -> int:
        """Record one status transition against the registry."""
        func = "status_changed"
        _dbg("OC_CHAT.MAIN", func, f"mirroring {from_status} -> {to_status}", 2)
 
        def _mirror(registry: object, actor_slot: str) -> None:
            from oc_chat.registry.domain import plan_set_status
 
            plan_set_status(
                registry,
                plan_id=plan_id,
                from_status=from_status,
                to_status=to_status,
                source="worker_self",
                note=note,
                actor_slot=actor_slot,
                actor_type="human_operator",
                command="plan_status set",
            )
 
        return self._run("plan.set_status", _mirror)
 
    def _run(
        self,
        operation_type: str,
        callback: Callable[[object, str], None],
    ) -> int:
        """Take the legacy-compat lock and run *callback*; report drift.
 
        Moved verbatim from ``plan_status._mirror_plan_status``.
        """
        func = "_run"
        try:
            context = self._context_factory()
            if context is None:
                return 0
            registry, _backend, actor_slot = context
            lock = registry.acquire_lock(
                "legacy-compat-write",
                actor_slot,
                60,
                {"source": "plan_status", "operation_type": operation_type},
            )
            try:
                callback(registry, actor_slot)
            finally:
                lock.release()
            return 0
        except Exception as exc:
            # EXTERNAL_ERROR: the registry is a separate store that may be
            # locked, migrating or absent; the worker's own status file is
            # already written, so this is reported, never raised.
            _dbg(
                "OC_CHAT.MAIN",
                func,
                f"EXTERNAL_ERROR: {operation_type}: {type(exc).__name__}: {exc}",
                2,
            )
            print(
                f"REGISTRY DRIFT - {operation_type} failed: {type(exc).__name__}: {exc}. "
                "Run `oct agent migrate verify`.",
                file=sys.stderr,
            )
            return 2
 
 
__all__ = [
    "CompatContext",
    "RegistryMirrorObserver",
    "compat_context",
    "registry_plan_exists",
    "registry_plan_row",
    "registry_plan_status",
]