From 2b79d68ba1ba8a553e4542808f9b8a8cf6c084a0 Mon Sep 17 00:00:00 2001 From: CAVIN <35207870+zzz163519@users.noreply.github.com> Date: Sat, 18 Jul 2026 13:08:01 +0800 Subject: [PATCH] feat(project-view): expose execution phase diagnostics --- lib/ccbd/project_view/service.py | 84 +++++++++++++++++ lib/maintenance_heartbeat/classifier.py | 9 +- test/test_ccbd_project_view.py | 115 ++++++++++++++++++++++++ test/test_maintenance_heartbeat.py | 25 ++++++ 4 files changed, 232 insertions(+), 1 deletion(-) diff --git a/lib/ccbd/project_view/service.py b/lib/ccbd/project_view/service.py index 3b83ada6c..2ce97a989 100644 --- a/lib/ccbd/project_view/service.py +++ b/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,6 +2002,8 @@ 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, @@ -1983,6 +2011,62 @@ def _comm_record( '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), + } } diff --git a/lib/maintenance_heartbeat/classifier.py b/lib/maintenance_heartbeat/classifier.py index 753a15203..ec984b852 100644 --- a/lib/maintenance_heartbeat/classifier.py +++ b/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 diff --git a/test/test_ccbd_project_view.py b/test/test_ccbd_project_view.py index b46a0b172..185589361 100644 --- a/test/test_ccbd_project_view.py +++ b/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() diff --git a/test/test_maintenance_heartbeat.py b/test/test_maintenance_heartbeat.py index 24dbe14e2..e06378864 100644 --- a/test/test_maintenance_heartbeat.py +++ b/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(