Expose execution phase and orphaned inbound diagnostics

lib/ccbd/project_view/service.py

@@ -190,12 +190,14 @@ class _CommsLookup:
     attempt_store: object | None
     reply_store: object | None
     message_store: object | None
+    inbound_store: object | None
     attempts_by_job_id: dict[str, object | None] = field(default_factory=dict)
     attempts_by_attempt_id: dict[str, object | None] = field(default_factory=dict)
     attempts_by_message_id: dict[tuple[str, str], object | None] = field(default_factory=dict)
     latest_attempt_by_message_agent: dict[tuple[str, str], object | None] = field(default_factory=dict)
     replies_by_reply_id: dict[str, object | None] = field(default_factory=dict)
     messages_by_message_id: dict[str, object | None] = field(default_factory=dict)
+    inbound_by_attempt: dict[tuple[str, str], object | None] = field(default_factory=dict)
 
     def attempt_by_job_id(self, job_id: object) -> object | None:
         key = str(job_id or '').strip()
@@ -259,6 +261,21 @@ def message_by_message_id(self, message_id: object) -> object | None:
             self.messages_by_message_id[key] = _call_store(self.message_store, 'get_latest', key)
         return self.messages_by_message_id[key]
 
+    def inbound_for_attempt(self, agent_name: object, attempt_id: object) -> object | None:
+        agent_key = str(agent_name or '').strip()
+        attempt_key = str(attempt_id or '').strip()
+        if not agent_key or not attempt_key:
+            return None
+        cache_key = (agent_key, attempt_key)
+        if cache_key not in self.inbound_by_attempt:
+            self.inbound_by_attempt[cache_key] = _call_store(
+                self.inbound_store,
+                'get_latest_for_attempt',
+                agent_key,
+                attempt_key,
+            )
+        return self.inbound_by_attempt[cache_key]
+
 
 @dataclass(frozen=True)
 class _CachedProjectViewResponse:
@@ -1493,6 +1510,7 @@ def _comms_view(
                     job,
                     reply_delivery=reply_delivery,
                     configured_agents=configured_agents,
+                    comms_lookup=comms_lookup,
                     running_recover_hint=running_recover_hint,
                     lineage_for_recoverability=lineage_for_recoverability,
                 ),
@@ -1575,6 +1593,7 @@ def _comms_lookup(dispatcher) -> _CommsLookup:
         attempt_store=getattr(control, '_attempt_store', None) if control is not None else None,
         reply_store=getattr(control, '_reply_store', None) if control is not None else None,
         message_store=getattr(control, '_message_store', None) if control is not None else None,
+        inbound_store=getattr(control, '_inbound_store', None) if control is not None else None,
     )
 
 
@@ -1946,6 +1965,7 @@ def _comm_record(
     *,
     reply_delivery,
     configured_agents: frozenset[str],
+    comms_lookup: _CommsLookup,
     running_recover_hint: str | None = None,
     lineage_for_recoverability=None,
 ) -> dict[str, object]:
@@ -1956,6 +1976,12 @@ def _comm_record(
     )
     if running_recover_hint and getattr(job, 'status', None) is JobStatus.RUNNING:
         business_status, status_label = 'blocked', 'stuck'
+    execution_phase, execution_phase_reason = _execution_phase(
+        job,
+        reply_delivery=reply_delivery,
+        configured_agents=configured_agents,
+        running_recover_hint=running_recover_hint,
+    )
     reply_status = reply_delivery.status.value if reply_delivery is not None else None
     updated_at = _comm_sort_key(job, reply_delivery)
     recoverability = comms_recoverability_for_job(
@@ -1976,13 +2002,71 @@ def _comm_record(
         'status': job.status.value,
         'business_status': business_status,
         'status_label': status_label,
+        'execution_phase': execution_phase,
+        'execution_phase_reason': execution_phase_reason,
         'body_preview': _body_preview(job.request.body),
         'reply_status': reply_status,
         'reply_delivery_job_id': reply_delivery.job_id if reply_delivery is not None else None,
         'chain': bool(getattr(job.request, 'reply_to', None)),
         'short_reason': _short_reason(job),
         **({'attachments': attachments} if attachments else {}),
         **recoverability.to_record(),
+        **(
+            _orphaned_active_inbound_evidence(dispatcher, job, comms_lookup=comms_lookup)
+            if execution_phase == 'orphaned'
+            else {}
+        ),
+    }
+
+
+def _execution_phase(
+    job,
+    *,
+    reply_delivery,
+    configured_agents: frozenset[str],
+    running_recover_hint: str | None,
+) -> tuple[str, str]:
+    if job.status in _COMMS_PENDING_STATUSES:
+        return 'queued', 'job_queued'
+    if job.status is JobStatus.RUNNING:
+        if running_recover_hint == 'provider_prompt_idle':
+            return 'orphaned', 'orphaned_active_inbound'
+        return 'executing', 'job_running'
+    if job.status is JobStatus.COMPLETED and _expects_reply_delivery(job, configured_agents):
+        if reply_delivery is None or reply_delivery.status in _COMMS_PENDING_STATUSES:
+            return 'reply_queued', 'reply_delivery_pending'
+        if reply_delivery.status is JobStatus.RUNNING:
+            return 'reply_delivering', 'reply_delivery_running'
+    return 'terminal', f'job_{job.status.value}'
+
+
+def _orphaned_active_inbound_evidence(
+    dispatcher,
+    job,
+    *,
+    comms_lookup: _CommsLookup,
+) -> dict[str, object]:
+    attempt = comms_lookup.attempt_by_job_id(getattr(job, 'job_id', None))
+    inbound = comms_lookup.inbound_for_attempt(
+        getattr(job, 'agent_name', None),
+        getattr(attempt, 'attempt_id', None),
+    )
+    control = getattr(dispatcher, '_message_bureau_control', None)
+    mailbox = _call_store(getattr(control, '_mailbox_store', None), 'load', getattr(job, 'agent_name', None))
+    lease = _call_store(getattr(control, '_lease_store', None), 'load', getattr(job, 'agent_name', None))
+    inbound_status = getattr(inbound, 'status', None)
+    mailbox_state = getattr(mailbox, 'mailbox_state', None)
+    lease_state = getattr(lease, 'lease_state', None)
+    return {
+        'orphaned_active_inbound': {
+            'inbound_event_id': getattr(inbound, 'inbound_event_id', None),
+            'inbound_status': getattr(inbound_status, 'value', inbound_status),
+            'attempt_id': getattr(attempt, 'attempt_id', None),
+            'mailbox_state': getattr(mailbox_state, 'value', mailbox_state),
+            'mailbox_head_inbound_event_id': getattr(mailbox, 'head_inbound_event_id', None),
+            'active_inbound_event_id': getattr(mailbox, 'active_inbound_event_id', None),
+            'lease_state': getattr(lease_state, 'value', lease_state),
+        }
     }
 
 

lib/maintenance_heartbeat/classifier.py

@@ -262,13 +262,20 @@ def _comms_issue(comm: Mapping[str, object]) -> dict[str, Any] | None:
             status=status,
         )
     if business_status in _CONCERN_COMMS_STATUSES:
+        execution_phase = _clean(comm.get('execution_phase'))
+        execution_reason = _clean(comm.get('execution_phase_reason'))
         return _issue(
             HEALTH_CONCERN,
             'comms',
             job_id=job_id,
             target=target,
-            reason=str(comm.get('block_reason') or business_status or 'comms_blocked'),
+            reason=(
+                execution_reason
+                if execution_phase == 'orphaned' and execution_reason
+                else str(comm.get('block_reason') or business_status or 'comms_blocked')
+            ),
             status=status,
+            execution_phase=execution_phase or None,
         )
     return None
 

test/test_ccbd_project_view.py

@@ -1738,6 +1738,29 @@ def test_project_view_returns_minimal_windows_agents_and_comms(tmp_path: Path) -
     assert [item['id'] for item in view['comms']] == ['job_running_1234', 'job_queued_5678']
     assert view['comms'][0]['sender'] == 'agent2'
     assert view['comms'][0]['target'] == 'agent1'
+    assert view['comms'][0]['execution_phase'] == 'executing'
+    assert view['comms'][1]['execution_phase'] == 'queued'
+
+
+def test_execution_phase_only_marks_request_anchored_provider_idle_as_orphaned() -> None:
+    job = _job('project', job_id='job_running', sender='agent2', target='agent1', status=JobStatus.RUNNING)
+    configured_agents = frozenset({'agent1', 'agent2'})
+
+    for hint in (None, 'provider_prompt_input_stuck', 'provider_prompt_idle_stale'):
+        phase = project_view_service._execution_phase(
+            job,
+            reply_delivery=None,
+            configured_agents=configured_agents,
+            running_recover_hint=hint,
+        )
+        assert phase == ('executing', 'job_running')
+
+    assert project_view_service._execution_phase(
+        job,
+        reply_delivery=None,
+        configured_agents=configured_agents,
+        running_recover_hint='provider_prompt_idle',
+    ) == ('orphaned', 'orphaned_active_inbound')
 
 
 def test_project_view_includes_provider_runtime_for_active_execution(tmp_path: Path) -> None:
@@ -2952,6 +2975,26 @@ def test_project_view_comms_marks_agent_reply_delivery_pending(tmp_path: Path) -
         updated_at='2026-05-20T12:00:01Z',
     )
     dispatcher._append_job(source)
+    delivering_source = _job(
+        project_id,
+        job_id='job_source_delivering',
+        sender='agent2',
+        target='agent3',
+        status=JobStatus.COMPLETED,
+        updated_at='2026-05-20T12:00:02Z',
+    )
+    dispatcher._append_job(delivering_source)
+    delivery = _reply_delivery_job(
+        project_id,
+        job_id='job_delivery_running',
+        source_agent='agent3',
+        source_job_id=delivering_source.job_id,
+        target='agent2',
+        status=JobStatus.RUNNING,
+        updated_at='2026-05-20T12:00:03Z',
+    )
+    dispatcher._append_job(delivery)
+    dispatcher._state.mark_active_for(TargetKind.AGENT, delivery.target_name, delivery.job_id)
     cmd_source = _job(
         project_id,
         job_id='job_cmd_source',
@@ -2987,6 +3030,11 @@ def test_project_view_comms_marks_agent_reply_delivery_pending(tmp_path: Path) -
     comms_by_id = {item['id']: item for item in service.build_response()['view']['comms']}
 
     assert comms_by_id[source.job_id]['business_status'] == 'delivering'
+    assert comms_by_id[source.job_id]['execution_phase'] == 'reply_queued'
+    assert comms_by_id[source.job_id]['execution_phase_reason'] == 'reply_delivery_pending'
+    assert comms_by_id[delivering_source.job_id]['business_status'] == 'delivering'
+    assert comms_by_id[delivering_source.job_id]['execution_phase'] == 'reply_delivering'
+    assert comms_by_id[delivering_source.job_id]['execution_phase_reason'] == 'reply_delivery_running'
     assert comms_by_id[cmd_source.job_id]['business_status'] == 'replied'
     assert comms_by_id[silent_source.job_id]['business_status'] == 'completed'
 
@@ -3981,6 +4029,8 @@ def test_project_view_marks_running_job_idle_after_provider_prompt_reappears(tmp
     assert comm['id'] == job.job_id
     assert comm['business_status'] == 'replying'
     assert comm['status_label'] == 'work'
+    assert comm['execution_phase'] == 'executing'
+    assert comm['execution_phase_reason'] == 'job_running'
     assert comm['recoverable'] is False
     assert comm['block_reason'] is None
 
@@ -4041,10 +4091,75 @@ def test_project_view_does_not_mark_fresh_running_prompt_idle_as_recoverable(tmp
     assert comm['id'] == job.job_id
     assert comm['business_status'] == 'replying'
     assert comm['status_label'] == 'work'
+    assert comm['execution_phase'] == 'executing'
+    assert comm['execution_phase_reason'] == 'job_running'
     assert comm['recoverable'] is False
     assert comm['block_reason'] is None
 
 
+def test_project_view_marks_stale_claude_idle_job_as_orphaned(tmp_path: Path) -> None:
+    project_root = tmp_path / 'repo-claude-orphaned-inbound'
+    project_root.mkdir()
+    layout = PathLayout(project_root)
+    project_id = compute_project_id(project_root)
+    base = _config()
+    config = replace(base, agents={**base.agents, 'agent3': _spec('agent3', 'claude')})
+    registry = AgentRegistry(layout, config)
+    registry.upsert(_runtime('agent3', project_id=project_id, state=AgentState.BUSY))
+    mount_manager = MountManager(layout, clock=lambda: NOW)
+    mount_manager.mark_mounted(project_id=project_id, pid=123, socket_path=layout.ccbd_socket_path, generation=1, started_at=NOW)
+    ProjectNamespaceStateStore(layout).save(
+        ProjectNamespaceState(
+            project_id=project_id,
+            namespace_epoch=3,
+            tmux_socket_path=str(layout.ccbd_tmux_socket_path),
+            tmux_session_name='ccb-claude-orphaned-inbound',
+            layout_version=2,
+        )
+    )
+    dispatcher = JobDispatcher(layout, config, registry, clock=lambda: NOW)
+    job_id = _submit(dispatcher, project_id, sender='agent1', target='agent3', body='stale running request')
+    dispatcher.tick()
+    job = dispatcher.get(job_id)
+    assert job is not None
+    job = replace(job, updated_at='2026-05-20T11:59:20Z')
+    dispatcher._append_job(job)
+    controller = ProjectNamespaceController(
+        layout,
+        project_id,
+        backend_factory=lambda socket_path=None: _ProviderIdleAfterRequestBackend(job.job_id),
+    )
+    service = ProjectViewService(
+        ProjectViewDependencies(
+            project_root=project_root,
+            project_id=project_id,
+            config=config,
+            registry=registry,
+            mount_manager=mount_manager,
+            namespace_state_store=ProjectNamespaceStateStore(layout),
+            dispatcher=dispatcher,
+            namespace_controller=controller,
+            clock=lambda: NOW,
+        )
+    )
+
+    comm = service.build_response()['view']['comms'][0]
+
+    assert comm['id'] == job.job_id
+    assert comm['business_status'] == 'blocked'
+    assert comm['execution_phase'] == 'orphaned'
+    assert comm['execution_phase_reason'] == 'orphaned_active_inbound'
+    assert comm['block_reason'] == 'provider_prompt_idle'
+    assert comm['recoverable'] is True
+    evidence = comm['orphaned_active_inbound']
+    assert evidence['inbound_event_id']
+    assert evidence['inbound_status'] == 'delivering'
+    assert evidence['mailbox_state'] == 'delivering'
+    assert evidence['mailbox_head_inbound_event_id'] == evidence['inbound_event_id']
+    assert evidence['active_inbound_event_id'] == evidence['inbound_event_id']
+    assert evidence['lease_state'] == 'acquired'
+
+
 def test_project_view_does_not_use_legacy_codex_prompt_idle_recovery_without_anchor(tmp_path: Path) -> None:
     project_root = tmp_path / 'repo-provider-idle-no-anchor'
     project_root.mkdir()

test/test_maintenance_heartbeat.py

@@ -174,6 +174,31 @@ def test_maintenance_classifier_keeps_active_ccb_job_healthy() -> None:
     assert evaluation.evidence == ()
 
 
+def test_maintenance_classifier_flags_orphaned_active_inbound() -> None:
+    evaluation = evaluate_project_view(
+        _project_view_payload(
+            agent_state='pending',
+            agent_reason='provider_prompt_idle',
+            current_job_id='job_orphaned',
+            queue_depth=2,
+            comms=(
+                {
+                    'id': 'job_orphaned',
+                    'target': 'demo',
+                    'business_status': 'blocked',
+                    'status': 'running',
+                    'execution_phase': 'orphaned',
+                    'execution_phase_reason': 'orphaned_active_inbound',
+                },
+            ),
+        )
+    )
+
+    assert evaluation.health == 'concern'
+    assert evaluation.summary['concern_comms_count'] == 1
+    assert any(item.get('reason') == 'orphaned_active_inbound' for item in evaluation.evidence)
+
+
 def test_maintenance_classifier_keeps_active_comms_without_current_job_healthy() -> None:
     evaluation = evaluate_project_view(
         _project_view_payload(