diff --git a/src/bg_monitor.py b/src/bg_monitor.py index 4ff437043..28f7f2225 100644 --- a/src/bg_monitor.py +++ b/src/bg_monitor.py @@ -157,6 +157,12 @@ async def _run_followup(rec: dict) -> FollowupResult: context.append(_background_result_message(rec)) authority = authority.restrict(disabled_tools=get_setting("disabled_tools", []) or ()) full, tool_events = await _drain_agent(sess, context, request_authority=authority) + # An awaited continuation must not deliver a result after its immutable + # linkage disappears or is replaced. This check grants no new authority. + try: + validate_job(resource) + except (ValueError, TypeError, OSError, RuntimeError): + return FollowupResult.TERMINAL_UNFOLLOWABLE # Persist ONLY the assistant continuation so it renders as a normal agent # turn — a standard chat bubble plus `tool_events` that the frontend @@ -184,7 +190,8 @@ async def _process_followup(rec): from src.agent_runtime.process_resources import job_from_record bg_jobs.mark_followed_up(rec["id"], expected=job_from_record(rec)) elif outcome is FollowupResult.TERMINAL_UNFOLLOWABLE: - bg_jobs.mark_unfollowable(rec["id"], expected_record=rec) + if not bg_jobs.mark_unfollowable(rec["id"], expected_record=rec): + return FollowupResult.RETRYABLE_LATER logger.warning("bg-followup: job %s has no valid continuation linkage; retired from pending", rec.get("id")) return outcome diff --git a/tests/test_wave3_background_followup.py b/tests/test_wave3_background_followup.py index 50bf02789..4deeb0ef5 100644 --- a/tests/test_wave3_background_followup.py +++ b/tests/test_wave3_background_followup.py @@ -82,3 +82,23 @@ def test_stale_terminal_snapshot_cannot_suppress_new_generation(store): assert not bg_jobs.mark_unfollowable('job', expected_record=old) assert 'followup_state' not in bg_jobs.peek('job') assert resources.launch_path(new.generation).exists() + + +async def test_linkage_lost_during_continuation_cannot_deliver(store, monkeypatch, monitor_session): + _, rec = seed(store, status='done') + async def interrupted(*args, **kwargs): + (bg_jobs._JOBS_DIR / 'job.authority.json').unlink() + return 'must not be delivered', [] + monkeypatch.setattr(bg_monitor, '_drain_agent', interrupted) + assert await bg_monitor._process_followup(rec) is bg_monitor.FollowupResult.TERMINAL_UNFOLLOWABLE + assert not monitor_session + assert not bg_jobs.pending_followups() + + +async def test_stale_terminal_outcome_retries_current_record(store, monkeypatch, monitor_session): + _, old = seed(store, status='done') + seed(store, status='done') + async def terminal(rec): return bg_monitor.FollowupResult.TERMINAL_UNFOLLOWABLE + monkeypatch.setattr(bg_monitor, '_run_followup', terminal) + assert await bg_monitor._process_followup(old) is bg_monitor.FollowupResult.RETRYABLE_LATER + assert bg_jobs.pending_followups()