runtime: expose retryable finalizer runs

docs/development/02-session-lifecycle.md

@@ -259,6 +259,11 @@ tasks opt into the same result mode when their caller must preserve raised
 cleanup errors. Callers can inspect or pop those results and use `unwrap()` to
 apply their existing exception-precedence policy.
 
+`RuntimeScope.run_finalizer()` can invoke a registered finalizer before the
+root close begins without closing ordinary task admission. Concurrent callers
+share one attempt, a successful attempt is not repeated by the later root
+close, and a failed attempt remains visible and retryable.
+
 ```mermaid
 stateDiagram-v2
     [*] --> Scheduled

src/easycat/runtime/scope.py

@@ -489,6 +489,21 @@ def add_finalizer(
             factory=factory,
         )
 
+    async def run_finalizer(self, name: str) -> None:
+        """Run one registered finalizer without closing ordinary admission.
+
+        Concurrent callers join the same attempt. A successful attempt is not
+        repeated by a later caller or :meth:`close`; a failed attempt retains
+        its terminal result and the next call retries the registered factory.
+        """
+        if not name:
+            raise ValueError("RuntimeScope finalizer name must be non-empty")
+        self.root._require_open()
+        node = self.root._finalizer_named(name)
+        if node is None:
+            raise ValueError(f"RuntimeScope finalizer {name!r} is not registered")
+        await self.root._run_finalizer(node)
+
     def terminal_results(self, name: str | None = None) -> tuple[RuntimeTerminalResult, ...]:
         """Return retained task and finalizer results across this subtree."""
         return tuple(

tests/runtime/test_scope.py

@@ -1067,6 +1067,98 @@ async def test_runtime_scope_finalizer_registration_rejects_duplicates_and_cohor
     await other.cancel_and_drain()
 
 
+@pytest.mark.asyncio
+async def test_run_finalizer_shares_attempt_without_closing_task_admission() -> None:
+    root = _attached_root("session")
+    child = root.create_child("provider")
+    started = asyncio.Event()
+    release = asyncio.Event()
+    calls = 0
+
+    async def cleanup() -> str:
+        nonlocal calls
+        calls += 1
+        started.set()
+        await release.wait()
+        return "closed"
+
+    child.add_finalizer("provider-close", cleanup)
+    first = asyncio.create_task(root.run_finalizer("provider-close"))
+    await started.wait()
+    second = asyncio.create_task(child.run_finalizer("provider-close"))
+    await asyncio.sleep(0)
+
+    assert calls == 1
+
+    release.set()
+    await asyncio.gather(first, second)
+
+    admitted = child.create_task("after-finalize", asyncio.sleep(0))
+    await child.drain("after-finalize")
+    assert admitted.done()
+    assert root.state is RuntimeScopeState.OPEN
+    results = root.terminal_results("provider-close")
+    assert len(results) == 1
+    assert results[0].status is RuntimeResultStatus.COMPLETED
+    assert results[0].unwrap() == "closed"
+
+    assert await root.close() is RuntimeScopeState.CLOSED
+    assert calls == 1
+
+
+@pytest.mark.asyncio
+async def test_run_finalizer_retains_failure_and_retries_factory() -> None:
+    root = _attached_root("session")
+    attempts = 0
+
+    async def cleanup() -> None:
+        nonlocal attempts
+        attempts += 1
+        if attempts == 1:
+            raise RuntimeError("provider close failed")
+
+    root.add_finalizer("provider-close", cleanup)
+
+    with pytest.raises(RuntimeError, match="provider close failed"):
+        await root.run_finalizer("provider-close")
+
+    assert root.state is RuntimeScopeState.OPEN
+    assert root.terminal_results("provider-close")[0].status is RuntimeResultStatus.RAISED
+
+    await root.run_finalizer("provider-close")
+
+    assert attempts == 2
+    assert [result.status for result in root.terminal_results("provider-close")] == [
+        RuntimeResultStatus.RAISED,
+        RuntimeResultStatus.COMPLETED,
+    ]
+    assert await root.close() is RuntimeScopeState.CLOSED
+    assert attempts == 2
+
+
+@pytest.mark.asyncio
+async def test_run_finalizer_rejects_new_attempt_after_close_starts() -> None:
+    root = _attached_root("session")
+    release = asyncio.Event()
+    calls = 0
+
+    async def finalizer() -> None:
+        nonlocal calls
+        calls += 1
+
+    root.create_task("work", release.wait())
+    root.add_finalizer("provider-close", finalizer)
+    closing = asyncio.create_task(root.close(phases=("default", "provider-close")))
+    await asyncio.sleep(0)
+
+    with pytest.raises(RuntimeError, match="is closing"):
+        await root.run_finalizer("provider-close")
+
+    release.set()
+    assert await closing is RuntimeScopeState.CLOSED
+    assert calls == 1
+
+
 @pytest.mark.asyncio
 async def test_close_runs_finalizers_at_explicit_positions_between_cohorts() -> None:
     root = _attached_root("session")