Honour the worker's landing budget in the log capture drain

[ppXD]

Summary

• A capture whose observer was cancelled never returned: the live loop's only cancellation point was a poll wait, and Task.WhenAny hands back a cancelled Task.Delay instead of raising it, while the loop's other await (the source read) answers "nothing new" out of a length comparison with no token check. The loop spun at full speed and the worker tear-down never started, because it runs from the very cancellation the drain was refusing to observe (AgentRunLogCaptureBridge.CaptureLoopAsync).
• The final drain then had two ways to wait out its whole finalization budget while the host was going away — a destination refusing writes, and a source that had not sealed. It now reads the host's own ApplicationStopping (the same signal AgentRunExecutor's tear-down arm acts on, threaded through AgentRunLogCaptureOpenRequest.HostShutdown) and parks on either, leaving the row Open at its own fence for AgentRunLogCaptureRecoveryService to re-claim. CaptureStream.Parked is its own flag, distinct from Terminal, so nothing reads a park as a conclusion.
• The two parks get different grace, because they are not the same bet. A refusal parks at once — the destination said no and will say no again. An unsealed source gets one retry (SourceSealGracePasses, one PollInterval, 250 ms of a ten-second landing), because that wait is usually on a local copier draining the FIFO of an agent this drain just killed; parking it on the first pass would leave a capture whose every byte landed with no final-drain receipt, and the recovery sweep terminalizes such a stream rather than re-reading the spool. They also leave different durable state, deliberately: the refusal park lands on the remote_stall marker its own refusal wrote, while the unsealed-source park writes nothing — no destination refused anything, and a stall marker would send an operator hunting an incident that never happened. Both asymmetries are stated at the two sites.
• What waiting costs differs by which capture is draining, and both are stated where the claim is made: a session opened for the live run holds the job's token, so a drain that will not end is a landing that never STARTS; the session the tear-down itself opens to re-fold the dead agent holds ShutdownLeaseLandingBudget, so every second it spends is one the terminal write does not get.
• The drain's ceilings (operation timeout, finalization budget, transient-append backoff) move onto the injected TimeProvider; the poll cadence stays on the wall clock because it decides nothing. This removes a flake in Final_drain_retries_one_transient_append_and_preserves_the_complete_source, which raced a 300 ms wall budget against a 50 ms wall backoff. Production defaults are unchanged and pinned by a literal test.
• Both FakeLogSources stopped observing a cancellation token production does not observe: LocalProcessRunner.ReadAsync returns NoData where the fixtures threw, which is what kept this class of cancellation bug out of reach of the suite. The existing healthy-destination drain arm was likewise vacuous about capture — with no route the bridge fails every stream at open and hands back a passthrough session — so it now seeds a writable destination and is the integration witness for the live loop.

Test plan

• Unit: A_cancelled_observer_stops_the_capture_loop_even_though_the_source_never_looks_at_the_token — mutation (drop the loop's ThrowIfCancellationRequested) fails loudly at 20 s, as does Cancelled_observer_leaves_open_source_...
• Unit: A_drain_under_a_host_that_is_going_away_parks_on_the_first_refusal_... (virtual clock never moves ⇒ the drain waited for nothing) — mutation (remove the refusal park) fails loudly at 20 s
• Unit: A_drain_whose_source_never_seals_also_parks_for_a_host_that_is_going_away — mutation (remove the loop park) fails loudly at 20 s
• Unit: A_source_that_seals_one_poll_late_is_finalized_rather_than_parked_when_the_host_is_stopping — mutation (park consulted from the first pass) → no final-drain receipt, red
• Unit: A_drain_on_a_healthy_host_still_waits_out_the_same_refusal — the same refusal is still retried when no tear-down is in progress
• Unit: Final_drain_retries_one_transient_append_... on a FakeTimeProvider — exactly 2 attempts, inside the budget; mutation (budget 20 ms < backoff 50 ms) goes red
• Unit: The_deployed_capture_cadence_is_unchanged pins 250 ms / 5 s / 30 s / 50 ms / 100 ms / 1 s; mutation (poll 250 ms → 500 ms) goes red
• Integration: A_worker_shutting_down_lands_its_brokered_run_within_the_budget_even_when_the_log_destination_refuses_every_write — run lands Failed/model-credential-lease-lost in ~4 s, stdout stream Open at the run's fence with a stall marker; mutation (HostShutdown → None) leaves the run Running. Before the fixes: never returned (45 s bound). 3/3 consecutive
• Integration: A_run_that_finishes_while_the_host_is_stopping_parks_its_capture_instead_of_draining_to_the_budget — the observer returns normally with ApplicationStopping already raised; mutation (HostShutdown → None) exceeds the 20 s ceiling
• Integration: A_worker_shutting_down_ends_every_brokered_run_... now seeds a writable destination and captures for real (2 streams asserted); mutation (drop ThrowIfCancellationRequested) hangs it to its 120 s bound
• Full unit suite: 10919 passed, 1 skipped, 0 failed
• Integration: AgentRunExecutorTests (incl. the credential-broker arms) 109/109; and on the prior head AgentRunLogCompletionRecoveryAuditTests 32/32, AgentRunLogCaptureRecoveryFlowTests 18/18, AgentRunReattachFlowTests 13/13, AgentRunAbandonedLogCaptureFlowTests 12/12, AgentRunLogRuntimeTests 16/16, AgentRunLogRemoteStallFlowTests 8/8, AgentRunLogStorageFaultFlowTests 5/5, AgentRunLogProviderSwitchFlowTests 3/3, AgentRunLogLargeCaptureFlowTests 1/1