diff --git a/opc/layer2_organization/company_runtime.py b/opc/layer2_organization/company_runtime.py index 53f09f6..e1fb1c9 100644 --- a/opc/layer2_organization/company_runtime.py +++ b/opc/layer2_organization/company_runtime.py @@ -29,6 +29,8 @@ from opc.layer2_organization.phase import ( is_dispatchable, is_report_execution_work_item_metadata, is_review_execution_work_item_metadata, + is_runnable, + is_terminal, ) from opc.layer2_organization.metadata_ownership import sync_work_item_current_turn_mode from opc.layer2_organization.session_scoping import ( @@ -1266,11 +1268,41 @@ class CompanyRuntime: ) continue if session_status == "blocked" and not can_soft_wake: - _skip( - "session.status=blocked and no review-soft-wake entry in queue", - session=session_label, - ) - continue + # Stale-block reconcile: `blocked` means "parked on my focused + # item until something external advances it". The park reason + # is gone when that item is runnable again (rework bounce / + # children-done wake), when it is TERMINAL (the awaited event + # already happened — observed shape: a review-preempt turn + # leaves the session parked on its own approved review card), + # or when there is no focus at all. Phase is the single + # source of truth, so converge the session instead of + # skipping forever. + focused_id = str(session.focused_work_item_id or "").strip() + focused_item = work_item_map.get(focused_id) if focused_id else None + focused_phase = getattr(focused_item, "phase", None) + if not focused_id or ( + focused_item is not None + and (is_runnable(focused_phase) or is_terminal(focused_phase)) + ): + logger.info( + "claim reconcile: stale blocked session converged to " + "idle session={} focused={} phase={}", + session_label, + focused_id or "", + getattr(focused_phase, "value", focused_phase), + ) + session.current_task_id = "" + session.focused_work_item_id = "" + self._set_member_session_status(session, "idle") + session_status = "idle" + else: + _skip( + "session.status=blocked and no review-soft-wake entry in queue", + session=session_label, + focused=focused_id or None, + focused_phase=getattr(focused_phase, "value", None), + ) + continue role_session = self._role_session_for_member_session(session) role_session_status = "" if role_session is not None: diff --git a/tests/test_claim_blocked_session_reconcile.py b/tests/test_claim_blocked_session_reconcile.py new file mode 100644 index 0000000..e22afa2 --- /dev/null +++ b/tests/test_claim_blocked_session_reconcile.py @@ -0,0 +1,95 @@ +"""Regression: a blocked member session parked on a runnable work item must +converge to idle inside claim_runnable_tasks (OBS-8, t1/t3 native wedge). + +The wake write (children approved → parent READY) can race complete_claim's +review-preempt restore, leaving the in-memory session `blocked` while the DB +work item is runnable and unclaimed. The dispatcher then skips the session +every tick forever. The claim loop is the consumption point, so it owns the +final say: a blocked session whose park reason no longer exists is converged +to idle before the skip decision. +""" +from __future__ import annotations + +import unittest + +from opc.core.models import CompanyMemberSession, DelegationWorkItem, Phase +from opc.layer2_organization.company_runtime import CompanyRuntime + + +def _runtime() -> CompanyRuntime: + return CompanyRuntime(org_engine=None, communication=None, store=None) + + +def _session(role: str, focused: str) -> CompanyMemberSession: + session = CompanyMemberSession( + member_session_id=f"ms-{role}", + role_id=role, + employee_id=f"{role}-default", + ) + session.status = "blocked" + session.resident_status = "blocked" + session.focused_work_item_id = focused + return session + + +def _work_item(work_item_id: str, *, phase: Phase) -> DelegationWorkItem: + return DelegationWorkItem( + work_item_id=work_item_id, + run_id="r", + cell_id="c", + role_id="cmo", + seat_id="seat-cmo", + title="parent", + phase=phase, + claimed_by_role_runtime_session_id="", + ) + + +class BlockedSessionClaimReconcileTests(unittest.IsolatedAsyncioTestCase): + async def test_blocked_session_with_ready_focus_converges_to_idle(self) -> None: + runtime = _runtime() + session = _session("cmo", focused="wi-1") + runtime.member_sessions[session.member_session_id] = session + item = _work_item("wi-1", phase=Phase.READY) + + await runtime.claim_runnable_tasks([], work_items=[item]) + + self.assertEqual(session.status, "idle") + self.assertEqual(session.focused_work_item_id, "") + + async def test_blocked_session_with_empty_focus_converges_to_idle(self) -> None: + runtime = _runtime() + session = _session("cmo", focused="") + runtime.member_sessions[session.member_session_id] = session + + await runtime.claim_runnable_tasks([], work_items=[]) + + self.assertEqual(session.status, "idle") + + async def test_blocked_session_with_terminal_focus_converges_to_idle(self) -> None: + """Observed t3 wedge: preempt-restore left the session parked on its + own APPROVED review card — the awaited event already happened.""" + runtime = _runtime() + session = _session("cto", focused="review::wi-9::v1") + runtime.member_sessions[session.member_session_id] = session + item = _work_item("review::wi-9::v1", phase=Phase.APPROVED) + + await runtime.claim_runnable_tasks([], work_items=[item]) + + self.assertEqual(session.status, "idle") + self.assertEqual(session.focused_work_item_id, "") + + async def test_blocked_session_waiting_children_stays_blocked(self) -> None: + runtime = _runtime() + session = _session("cmo", focused="wi-1") + runtime.member_sessions[session.member_session_id] = session + item = _work_item("wi-1", phase=Phase.WAITING_FOR_CHILDREN) + + await runtime.claim_runnable_tasks([], work_items=[item]) + + self.assertEqual(session.status, "blocked") + self.assertEqual(session.focused_work_item_id, "wi-1") + + +if __name__ == "__main__": + unittest.main()