fix(runtime): revalidate background linkage before delivery

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
 

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()