Classes
class	RegistryMirrorObserver

Functions
tuple[object, object, str]\|None	compat_context ()
bool	registry_plan_exists (object registry, str plan_id)
str\|None	registry_plan_status (object registry, str plan_id)
object\|None	registry_plan_row (object registry, str plan_id)

Variables
	CompatContext = Callable[[], "tuple[object, object, str] \| None"]

Detailed Description

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)

Function Documentation

◆ compat_context()

tuple[object, object, str] | None oc_chat.registry.mirror_observer.compat_context ( )

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.

Definition at line 75 of file mirror_observer.py.

◆ registry_plan_exists()

bool oc_chat.registry.mirror_observer.registry_plan_exists	(	object	registry,
		str	plan_id )

True when *plan_id* already has a row in the registry.

Definition at line 106 of file mirror_observer.py.

Here is the call graph for this function:

◆ registry_plan_row()

object \| None oc_chat.registry.mirror_observer.registry_plan_row	(	object	registry,
		str	plan_id )

The registry's ``plans`` row for *plan_id*, or ``None``.

Definition at line 117 of file mirror_observer.py.

Here is the caller graph for this function:

◆ registry_plan_status()

str \| None oc_chat.registry.mirror_observer.registry_plan_status	(	object	registry,
		str	plan_id )

The registry's current status for *plan_id*, or ``None``.

Definition at line 111 of file mirror_observer.py.

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

◆ CompatContext

oc_chat.registry.mirror_observer.CompatContext = Callable[[], "tuple[object, object, str] | None"]

Definition at line 72 of file mirror_observer.py.

Classes

Functions

Variables