diff --git a/src/skulk/api/main.py b/src/skulk/api/main.py index 97e4a0c3..255a8ed6 100644 --- a/src/skulk/api/main.py +++ b/src/skulk/api/main.py @@ -1567,6 +1567,11 @@ def __init__( # this node), so an outstanding failed probe surfaces as `degraded` # instead of hiding until the third failure tears the steward down. self._steward_canary = StewardCanaryState() + # Live steward turns keyed by their advertised outer command id, so + # the generic cancel-by-id endpoint can stop a turn whose inner + # generation ids the caller never sees. Entries live exactly as long + # as their turn's chunk stream. + self._steward_turn_harnesses: dict[CommandId, StewardHarness] = {} self._event_log = DiskEventLog(_API_EVENT_LOG_DIR) if enable_event_log else None self._event_log_appends_since_retention_check = 0 self._system_id = SystemId() @@ -3371,10 +3376,27 @@ async def _steward_chat_completions( chunk_stream = self._extensions.tap_chat_stream( self._extension_context, task_params, chunk_stream ) + # The advertised command id must honor the generic cancel-by-id + # contract (POST /v1/cancel/{command_id}). The harness's inner + # generation ids are never shown to the caller, so the outer id maps + # to the harness for the turn's lifetime and cancel_command routes + # it there. Registration happens HERE, before the response exists: + # generate_chat_stream advertises the id in its SSE comment before + # it pulls the first chunk, so registering lazily inside the stream + # would leave a cancel-by-id 404 window during long prefill. + # Deregistration wraps the OUTERMOST response iterator: an inner + # layer (the keepalive wrapper emits bytes before pulling its + # source) can be abandoned before ever starting, and an unstarted + # generator's finally never runs — only the iterator Starlette + # itself drives is guaranteed to start and therefore to clean up. + self._steward_turn_harnesses[command_id] = harness if payload.stream: return StreamingResponse( - with_sse_keepalive( - generate_chat_stream(command_id, chunk_stream), + self._release_steward_turn_after( + command_id, + with_sse_keepalive( + generate_chat_stream(command_id, chunk_stream), + ), ), media_type="text/event-stream", headers={ @@ -3384,10 +3406,42 @@ async def _steward_chat_completions( }, ) return StreamingResponse( - collect_chat_response(command_id, chunk_stream), + self._release_steward_turn_after( + command_id, + collect_chat_response(command_id, chunk_stream), + ), media_type="application/json", ) + async def _release_steward_turn_after( + self, + command_id: CommandId, + response_stream: "AsyncIterator[str]", + ) -> "AsyncGenerator[str, None]": + """Deregister a steward turn from cancel-by-id when its response ends. + + The caller registers the turn BEFORE constructing the response (the + id is advertised to the client before the first chunk is pulled); + this wrapper owns only the removal. It must wrap the OUTERMOST + response iterator — the one Starlette drives — because that is the + only generator guaranteed to be started (and thus finalized) even + when the client disconnects after the first keepalive byte; an + abandoned inner generator that never started never runs its + ``finally``. + + Args: + command_id: The outer command id the caller registered. + response_stream: The fully assembled response body iterator. + + Yields: + The wrapped response's items, unchanged. + """ + try: + async for item in response_stream: + yield item + finally: + self._steward_turn_harnesses.pop(command_id, None) + async def _steward_extension_transform( self, history: list[StewardChatMessage], *, stream: bool ) -> tuple[list[StewardChatMessage], str, TextGenerationTaskParams]: @@ -3655,8 +3709,27 @@ async def delete_instance(self, instance_id: InstanceId) -> DeleteInstanceRespon instance_id=instance_id, ) - async def cancel_command(self, command_id: CommandId) -> CancelCommandResponse: - """Cancel an active command by closing its stream and notifying workers.""" + async def cancel_local_command(self, command_id: CommandId) -> bool: + """Cancel one locally registered command stream, if it exists. + + The single cancellation implementation shared by the public + cancel-by-id endpoint and the steward harness's turn cancellation: + closes the local response queue immediately (so the caller's stream + ends now, not when worker-side cancellation lands — a served engine + mid-generation may only observe it at completion) and notifies + workers through ``TaskCancelled``. + + Args: + command_id: The command whose local stream should be cancelled. + + Returns: + True when a local queue existed and was cancelled; False when the + command is unknown here or already completed. + + Side effects: + Sends ``TaskCancelled``, records the id so local stream cleanup + suppresses its ``TaskFinished``, and closes the queue. + """ sender = ( self._text_generation_queues.get(command_id) or self._image_generation_queues.get(command_id) @@ -3665,21 +3738,35 @@ async def cancel_command(self, command_id: CommandId) -> CancelCommandResponse: or self._audio_transcription_queues.get(command_id) ) if sender is None: - raise HTTPException( - status_code=404, - detail="Command not found or already completed", - ) - + return False await self._send(TaskCancelled(cancelled_command_id=command_id)) # Suppress the final TaskFinished emitted by local stream cleanup so the # worker can observe the Cancelled task and deliver runner-local cancel # before event-sourced task deletion happens. self._cancelled_command_ids.add(command_id) sender.close() + return True - return CancelCommandResponse( - message="Command cancelled.", - command_id=command_id, + async def cancel_command(self, command_id: CommandId) -> CancelCommandResponse: + """Cancel an active command by closing its stream and notifying workers.""" + if await self.cancel_local_command(command_id): + return CancelCommandResponse( + message="Command cancelled.", + command_id=command_id, + ) + # A steward turn advertises one outer command id while its inner + # generations run under private ids; route the outer id to the + # harness so the generic cancel contract holds there too. + steward_turn = self._steward_turn_harnesses.get(command_id) + if steward_turn is not None: + await steward_turn.cancel_turn() + return CancelCommandResponse( + message="Steward turn cancelled.", + command_id=command_id, + ) + raise HTTPException( + status_code=404, + detail="Command not found or already completed", ) def _command_task_is_terminal(self, command_id: CommandId) -> bool: @@ -4646,15 +4733,27 @@ async def _expire_stale_vision_media(self, now: float) -> None: "Vision media timed out waiting for worker verification", ) - async def send_task_cancellation(self, command_id: CommandId) -> None: + async def send_task_cancellation( + self, command_id: CommandId, *, suppress_local_finish: bool = True + ) -> None: """Public cancellation seam for internal callers (the steward harness). Sends the same TaskCancelled command the HTTP cancel endpoint sends, - suppressing the final TaskFinished so the worker can observe the - cancelled task and stop the runner promptly. + by default suppressing the final TaskFinished so the worker can + observe the cancelled task and stop the runner promptly. + + Args: + command_id: The command to cancel on the workers. + suppress_local_finish: Record the id so local stream cleanup + skips its TaskFinished. Pass False when NO local stream will + ever open for this command (e.g. a cancellation accepted + before its chunk stream was created): the marker is only + discarded by stream finalization, so retaining it with no + stream would leak one set entry per cancellation. """ await self._send(TaskCancelled(cancelled_command_id=command_id)) - self._cancelled_command_ids.add(command_id) + if suppress_local_finish: + self._cancelled_command_ids.add(command_id) async def dispatch_text_generation( self, diff --git a/src/skulk/api/steward.py b/src/skulk/api/steward.py index daaf5348..0e9fd4ed 100644 --- a/src/skulk/api/steward.py +++ b/src/skulk/api/steward.py @@ -642,6 +642,10 @@ def __init__(self, api: "API") -> None: # abandoned stream (client disconnect, cancel button) can stop the # runner instead of leaving it generating for nobody. self._active_command_id: CommandId | None = None + # Latched by cancel_turn: the investigation loop checks it before + # every inner dispatch, so cancelling the advertised outer command + # stops the whole turn rather than just its current generation. + self._turn_cancelled = False # Aggregated across the turn's inner generations so the terminal # chunk reports real usage instead of null, and the model's actual # terminal reason (e.g. length) is preserved. @@ -713,7 +717,7 @@ async def canary_probe( # stream iteration, and the chunk stream's own cancellation handling # already sends TaskCancelled and finalizes; a second cancel here # would just duplicate it for an already-finalized command. - with anyio.move_on_after(CANARY_PROBE_TIMEOUT_SECONDS): + with anyio.move_on_after(CANARY_PROBE_TIMEOUT_SECONDS) as deadline_scope: async for chunk in chunk_stream: if isinstance(chunk, ErrorChunk): return False @@ -725,6 +729,12 @@ async def canary_probe( got_text = True if isinstance(chunk, TokenChunk) and chunk.finish_reason is not None: break + # A cancelled deadline is a failed probe no matter what arrived + # first: partial output followed by a stall is exactly the wedge + # this canary exists to catch, and counting it as success would + # reset the failure run the teardown threshold depends on. + if deadline_scope.cancelled_caught: + return False return got_text def steward_instance(self) -> tuple[InstanceId, str] | None: @@ -952,6 +962,37 @@ async def run_turn_chunks( with anyio.move_on_after(2, shield=True): await self._api.send_task_cancellation(abandoned) + async def cancel_turn(self) -> None: + """Cancel the running turn on behalf of the advertised command id. + + The chat surface advertises one outer command id for the whole turn + while the harness dispatches per-step inner generations under ids the + client never sees. Cancelling the outer id must therefore stop the + inner generation currently in flight AND latch the turn closed so + the investigation loop does not simply dispatch its next step. + + Side effects: + Cancels the active inner command through the API's shared local + cancellation path (closing its local queue so this turn's stream + ends immediately), falling back to a bare worker notification + when the queue is already gone. + """ + self._turn_cancelled = True + active = self._active_command_id + if active is None: + return + self._active_command_id = None + # Close the inner command's LOCAL queue, not just notify workers: a + # served engine mid-generation may observe worker-side cancellation + # only at completion, and the accepted cancel must end this response + # now, not then. When the queue is already gone there is no stream + # left to finalize, so the fallback must not retain the local-finish + # marker (nothing would ever discard it). + if not await self._api.cancel_local_command(active): + await self._api.send_task_cancellation( + active, suppress_local_finish=False + ) + async def _run_investigation( self, messages: list[ChatCompletionMessage], @@ -962,6 +1003,8 @@ async def _run_investigation( wrap it with abandonment cleanup.""" reply = "" for step_index in range(MAX_STEPS_PER_TURN): + if self._turn_cancelled: + return if step_index == MAX_STEPS_PER_TURN - 1: messages.append( ChatCompletionMessage( @@ -1128,6 +1171,19 @@ async def _generate_events( task_params, target_instance_id=instance_id ) self._active_command_id = command.command_id + if self._turn_cancelled: + # cancel_turn ran while this dispatch was in flight: the latch + # was set before this inner id existed, so cancel_turn had + # nothing to stop. Honor it here rather than letting an + # accepted cancellation stream one more full generation. The + # step ends without a result event, which the loop reads as an + # empty final answer. No chunk stream ever opens for this + # command, so the local-finish marker must not be retained. + self._active_command_id = None + await api.send_task_cancellation( + command.command_id, suppress_local_finish=False + ) + return # No extension tap: this is one investigation step, not the turn. # The turn's single tap is applied by the caller of # run_turn_chunks (API._steward_chat_completions), so observers see diff --git a/src/skulk/api/tests/test_cancel_command.py b/src/skulk/api/tests/test_cancel_command.py index 112861c0..c213907d 100644 --- a/src/skulk/api/tests/test_cancel_command.py +++ b/src/skulk/api/tests/test_cancel_command.py @@ -42,6 +42,7 @@ def _make_api() -> Any: api._vision_media_failures = {} # pyright: ignore[reportPrivateUsage] api._vision_media_packet_sender = None # pyright: ignore[reportPrivateUsage] api._cancelled_command_ids = set() # pyright: ignore[reportPrivateUsage] + api._steward_turn_harnesses = {} # pyright: ignore[reportPrivateUsage] api._chunk_reorder = {} # pyright: ignore[reportPrivateUsage] api._data_dedup_cursor = {} # pyright: ignore[reportPrivateUsage] api._data_plane_observer = DataPlaneObserver( # pyright: ignore[reportPrivateUsage] @@ -188,3 +189,22 @@ async def test_finalize_command_stream_reports_natural_completion() -> None: task_finished = api._send.call_args[0][0] assert task_finished.finished_command_id == cid assert cid not in queue + + +@pytest.mark.asyncio +async def test_send_task_cancellation_without_stream_retains_no_marker() -> None: + """A no-stream cancellation must not leak a local-finish marker. + + The marker is discarded only by stream finalization; a cancellation for + a command whose chunk stream never opens (the steward dispatch race) + would otherwise grow `_cancelled_command_ids` forever (#833). + """ + api = _make_api() + cid = CommandId("no-stream-cmd") + + await api.send_task_cancellation(cid, suppress_local_finish=False) + assert cid not in api._cancelled_command_ids + api._send.assert_called_once() + + await api.send_task_cancellation(cid) + assert cid in api._cancelled_command_ids diff --git a/src/skulk/api/tests/test_steward_canary.py b/src/skulk/api/tests/test_steward_canary.py index 16596e67..981da28c 100644 --- a/src/skulk/api/tests/test_steward_canary.py +++ b/src/skulk/api/tests/test_steward_canary.py @@ -1,9 +1,18 @@ -"""Canary target selection: pure decision logic for the liveness probe.""" +"""Canary behavior: probe target selection and probe outcome semantics.""" -from skulk.api.steward import canary_probe_target +from collections.abc import AsyncGenerator +from types import SimpleNamespace +from typing import TYPE_CHECKING, cast + +import anyio +import pytest + +import skulk.api.steward as steward_module +from skulk.api.steward import StewardHarness, canary_probe_target from skulk.master.placement import place_instance from skulk.master.tests.test_placement import fully_connected_three_nodes from skulk.shared.models.model_cards import ModelCard, ModelId, ModelTask +from skulk.shared.types.chunks import TokenChunk from skulk.shared.types.commands import PlaceInstance from skulk.shared.types.common import CommandId from skulk.shared.types.memory import Memory @@ -13,6 +22,9 @@ from skulk.shared.types.worker.runners import RunnerLoading, RunnerReady, RunnerRunning from skulk.shared.types.worker.shards import Sharding +if TYPE_CHECKING: + from skulk.api.main import API + def _steward_placement() -> dict[InstanceId, Instance]: topology, node_memory, node_network, _node_ids = fully_connected_three_nodes( @@ -179,3 +191,63 @@ def test_terminal_lifecycle_tasks_do_not_mute_the_canary() -> None: assert ( canary_probe_target(placed, runners, tasks, host) == instance.instance_id ) + + +class _WedgedStreamApi: + """Stub API whose generation emits text and then stalls forever.""" + + def __init__(self, model_id: str) -> None: + self._model_id = model_id + + async def running_model_card(self, model_id: ModelId) -> ModelCard: + return ModelCard( + model_id=ModelId(self._model_id), + storage_size=Memory.from_gb(3), + n_layers=12, + hidden_size=30, + supports_tensor=True, + tasks=[ModelTask.TextGeneration], + ) + + async def dispatch_text_generation( + self, + task_params: TextGenerationTaskParams, + target_instance_id: InstanceId | None = None, + ) -> object: + return SimpleNamespace(command_id=CommandId()) + + def text_generation_chunk_stream( + self, + command: object, + task_params: TextGenerationTaskParams, + *, + extension_tap: bool = True, + ) -> "AsyncGenerator[TokenChunk, None]": + async def _stream() -> "AsyncGenerator[TokenChunk, None]": + yield TokenChunk( + model=ModelId(self._model_id), + text="O", + token_id=-1, + usage=None, + finish_reason=None, + ) + # The partial-output wedge: text arrived, the terminal never does. + await anyio.Event().wait() + + return _stream() + + +async def test_partial_output_then_stall_is_a_failed_probe( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cancelled deadline must fail the probe even after text arrived. + + Counting the partial output as success would clear the consecutive + failure run, so the exact wedge the canary exists to catch could never + reach the three-failure teardown threshold (#830). + """ + monkeypatch.setattr(steward_module, "CANARY_PROBE_TIMEOUT_SECONDS", 0.05) + api = _WedgedStreamApi("canary-brain") + harness = StewardHarness(cast("API", cast(object, api))) + + assert await harness.canary_probe(InstanceId(), "canary-brain") is False diff --git a/src/skulk/api/tests/test_steward_chunk_stream.py b/src/skulk/api/tests/test_steward_chunk_stream.py index b32cb681..68093ff7 100644 --- a/src/skulk/api/tests/test_steward_chunk_stream.py +++ b/src/skulk/api/tests/test_steward_chunk_stream.py @@ -4,6 +4,7 @@ tool collaborators is the loop's unit-test seam. """ +from collections.abc import AsyncGenerator from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: @@ -206,3 +207,195 @@ async def test_complete_literal_example_survives_in_final_answer() -> None: token_chunks = [c for c in chunks if isinstance(c, TokenChunk)] content = "".join(c.text for c in token_chunks if not c.is_thinking) assert content == answer + + +async def test_cancel_turn_stops_the_generation_and_the_loop() -> None: + """Cancelling the advertised outer id ends the whole turn (#830). + + The latch matters as much as the cancellation: without it the + investigation loop would treat the cancelled step as complete and + dispatch the next one, leaving the turn running behind the caller's + back. + """ + cancelled: list[object] = [] + + class _Api: + async def cancel_local_command(self, command_id: object) -> bool: + # The shared path closes the local queue; recording it here is + # the test's proof the turn cancelled through that path. + cancelled.append(command_id) + return True + + async def send_task_cancellation(self, command_id: object) -> None: + raise AssertionError( + "the fallback must not fire when the local cancel succeeded" + ) + + # A script that would otherwise tool-loop for the full step budget. + harness = _ScriptedHarness(turns=[("", [_call("get_cluster_state")])]) + harness._api = cast("API", cast(object, _Api())) # pyright: ignore[reportPrivateUsage] + harness._active_command_id = cast(Any, "cmd-inner-1") # pyright: ignore[reportPrivateUsage] + + stream = harness.run_turn_chunks( + [StewardChatMessage(role="user", content="hi")] + ) + first = await stream.__anext__() + assert isinstance(first, TokenChunk) + generations_before_cancel = len(harness.system_prompts) + + await harness.cancel_turn() + assert cancelled == ["cmd-inner-1"] + + async for _chunk in stream: + pass + # The loop stopped at the latch instead of dispatching further steps. + assert len(harness.system_prompts) == generations_before_cancel + + +async def test_cancel_command_routes_a_registered_steward_turn() -> None: + """POST /v1/cancel/{outer_id} must reach the turn's harness, not 404.""" + from skulk.api.main import API + from skulk.shared.types.common import CommandId + + cancelled: list[object] = [] + + class _Api: + async def cancel_local_command(self, command_id: object) -> bool: + # The inner queue is already gone in this scenario; the harness + # must fall back to the bare worker notification. + return False + + async def send_task_cancellation( + self, command_id: object, *, suppress_local_finish: bool = True + ) -> None: + # The queue is already gone, so nothing would ever discard a + # retained local-finish marker (#833). + assert suppress_local_finish is False + cancelled.append(command_id) + + harness = _ScriptedHarness(turns=[("irrelevant", [])]) + harness._api = cast("API", cast(object, _Api())) # pyright: ignore[reportPrivateUsage] + harness._active_command_id = cast(Any, "cmd-inner-9") # pyright: ignore[reportPrivateUsage] + + outer_id = CommandId() + api = API.__new__(API) + api._text_generation_queues = {} # pyright: ignore[reportPrivateUsage] + api._image_generation_queues = {} # pyright: ignore[reportPrivateUsage] + api._embedding_queues = {} # pyright: ignore[reportPrivateUsage] + api._audio_speech_queues = {} # pyright: ignore[reportPrivateUsage] + api._audio_transcription_queues = {} # pyright: ignore[reportPrivateUsage] + api._steward_turn_harnesses = {outer_id: harness} # pyright: ignore[reportPrivateUsage] + + response = await api.cancel_command(outer_id) + + assert response.command_id == outer_id + assert cancelled == ["cmd-inner-9"] + + +async def test_cancellation_racing_dispatch_still_cancels_the_step() -> None: + """A latch set while dispatch is in flight cancels the fresh command. + + cancel_turn can run between step start and dispatch completion, when no + inner id exists yet to cancel. The step must then cancel the command it + just obtained and end the turn instead of streaming one more full + generation behind an accepted cancellation (#833). + """ + from types import SimpleNamespace + + from skulk.shared.models.model_cards import ModelCard, ModelTask + from skulk.shared.types.memory import Memory + + cancelled: list[object] = [] + stream_requests: list[object] = [] + harness_holder: list[StewardHarness] = [] + + class _RacingApi: + async def running_model_card(self, model_id: ModelId) -> ModelCard: + return ModelCard( + model_id=ModelId("steward-brain"), + storage_size=Memory.from_gb(3), + n_layers=12, + hidden_size=30, + supports_tensor=True, + tasks=[ModelTask.TextGeneration], + ) + + async def dispatch_text_generation( + self, task_params: object, target_instance_id: object = None + ) -> object: + # The race: cancellation lands while dispatch is in flight. + await harness_holder[0].cancel_turn() + return SimpleNamespace(command_id="cmd-fresh-inner") + + def text_generation_chunk_stream( + self, command: object, task_params: object, *, extension_tap: bool = True + ) -> object: + stream_requests.append(command) + raise AssertionError("a cancelled step must not open a stream") + + async def send_task_cancellation( + self, command_id: object, *, suppress_local_finish: bool = True + ) -> None: + # The race path must not retain a local-finish marker: no chunk + # stream ever opens for this command (#833). + assert suppress_local_finish is False + cancelled.append(command_id) + + harness = StewardHarness(cast("API", cast(object, _RacingApi()))) + harness_holder.append(harness) + harness.steward_instance = lambda: (InstanceId(), "steward-brain") + + chunks = [ + chunk + async for chunk in harness.run_turn_chunks( + [StewardChatMessage(role="user", content="hi")] + ) + ] + + assert cancelled == ["cmd-fresh-inner"] + assert stream_requests == [] + final = chunks[-1] + assert isinstance(final, TokenChunk) + assert final.finish_reason == "stop" + + +async def test_release_wrapper_deregisters_on_early_disconnect() -> None: + """An early disconnect must still deregister the turn (#833). + + The keepalive layer emits its first byte before pulling its source, so + a client can vanish while every inner generator is still unstarted; an + unstarted generator's ``finally`` never runs. The release wrapper sits + outermost — the iterator Starlette itself drives — so it is always + started and its cleanup always fires, even when the wrapped source + never ran at all. + """ + from skulk.api.main import API + from skulk.shared.types.common import CommandId + + api = API.__new__(API) + outer_id = CommandId() + harness = _ScriptedHarness(turns=[("irrelevant", [])]) + api._steward_turn_harnesses = {outer_id: harness} # pyright: ignore[reportPrivateUsage] + + source_started: list[bool] = [] + + async def _source() -> "AsyncGenerator[str, None]": + source_started.append(True) + yield "never reached" + + async def _keepalive_like( + source: "AsyncGenerator[str, None]", + ) -> "AsyncGenerator[str, None]": + yield ": keep-alive\n\n" + async for item in source: + yield item + + stream = api._release_steward_turn_after( # pyright: ignore[reportPrivateUsage] + outer_id, _keepalive_like(_source()) + ) + first = await stream.__anext__() + assert first == ": keep-alive\n\n" + await stream.aclose() + + assert outer_id not in api._steward_turn_harnesses # pyright: ignore[reportPrivateUsage] + assert source_started == [], "the inner source never started, by design" diff --git a/src/skulk/master/main.py b/src/skulk/master/main.py index 3994fcef..e2339c00 100644 --- a/src/skulk/master/main.py +++ b/src/skulk/master/main.py @@ -28,13 +28,22 @@ usable_vram_by_node, ) from skulk.shared.apply import apply +from skulk.shared.backends import engine_of, platform_compatible_backends from skulk.shared.constants import SKULK_EVENT_LOG_DIR, SKULK_TRACING_ENABLED from skulk.shared.log_summaries import summarize_command_for_log +from skulk.shared.models.capabilities import resolve_model_capability_profile from skulk.shared.models.memory_estimate import ( estimate_shard_footprint, shard_fraction_of_model, ) -from skulk.shared.models.model_cards import ModelId, get_card, get_model_cards +from skulk.shared.models.model_cards import ( + ModelCard, + ModelId, + ModelTask, + card_serves_speech, + get_card, + get_model_cards, +) from skulk.shared.types.commands import ( AddCustomModelCard, AudioTranscription, @@ -235,6 +244,107 @@ def observe(self, *, now: float, idle: bool) -> float | None: NODE_HEARTBEAT_GAP_WARNING = timedelta(seconds=10) +def steward_candidate_is_servable(card: "ModelCard") -> bool: + """Whether a configured steward candidate can actually serve steward turns. + + The steward harness always dispatches ``TextGeneration`` with server-side + tools, so a candidate must be a text-generation card whose resolved + capability profile supports tool calling. Anything else (an embedding, + image, speech, or tool-less card named through an operator override of + ``steward_models``) would place a steward that fails every turn instead + of falling through to the next candidate. The bundled defaults are + additionally CI-locked to pass this check. + + Args: + card: The candidate's model card. + + Returns: + True when the card can serve tool-driven steward text generation. + """ + if ModelTask.TextGeneration not in card.tasks: + return False + # Routing truth: worker bootstrap dispatches image, embedding, and speech + # cards to their specialized runners BEFORE text-engine selection, so a + # multi-task card carrying any of those never reaches a text runner no + # matter what else it declares. + if ( + ModelTask.TextToImage in card.tasks + or ModelTask.ImageToImage in card.tasks + or ModelTask.TextEmbedding in card.tasks + or card_serves_speech(card) + ): + return False + profile = resolve_model_capability_profile(card.model_id, model_card=card) + if not profile.supports_tool_calling: + return False + # Backend truth, not just model truth: a card whose only platform- + # servable engine is vllm needs an explicit tool-call parser pin, or + # the launched server rejects every tools-bearing request and the + # steward would place but fail every turn. + servable_engines = { + engine + for tag in platform_compatible_backends( + card.placement.compatible_backends, + card_serves_vision=card.vision is not None, + card_serves_speech=False, + ) + if (engine := engine_of(tag)) is not None + } + return not ( + servable_engines == {"vllm"} + and (card.runtime is None or card.runtime.vllm_tool_call_parser is None) + ) + + +def placement_may_select_parserless_vllm( + card: "ModelCard", placements: "Mapping[InstanceId, Instance]" +) -> bool: + """Whether a minted placement could serve vllm for a card with no parser. + + The pre-placement servability gate can only reject a card whose EVERY + servable engine needs the parser; a multi-engine card passes it and may + still end up on vllm on the fleet at hand. This checks the backends + placement actually stamped, so the walk can skip the candidate instead + of committing a steward whose server rejects every tools-bearing + request. An UNSTAMPED shard (``resolved_backend=None``, the telemetry + warm-up window) counts as selectable too: the worker then falls back to + its local probe, which is free to pick vllm, so the only safe answer + while vllm is among the card's servable engines is "not yet" — the + invariant simply retries on a later tick once resources have arrived. + + Args: + card: The candidate's model card. + placements: ONLY the instances minted by this placement. The caller + must filter out pre-existing state (``place_instance`` returns + existing instances plus the new one), or an unrelated vllm or + unstamped shard would falsely condemn the candidate. + + Returns: + True when the card pins no ``runtime.vllm_tool_call_parser``, vllm + is among its platform-servable engines, and any shard either + resolved to vllm or carries no stamped backend. + """ + if card.runtime is not None and card.runtime.vllm_tool_call_parser is not None: + return False + servable_engines = { + engine + for tag in platform_compatible_backends( + card.placement.compatible_backends, + card_serves_vision=card.vision is not None, + card_serves_speech=False, + ) + if (engine := engine_of(tag)) is not None + } + if "vllm" not in servable_engines: + return False + for instance in placements.values(): + for shard in instance.shard_assignments.runner_to_shard.values(): + backend: str | None = getattr(shard, "resolved_backend", None) + if backend is None or engine_of(backend) == "vllm": + return True + return False + + def _aware_timestamp(when: datetime) -> datetime: """Return a timestamp that is safe to compare with UTC receipt times.""" return when if when.tzinfo is not None else when.replace(tzinfo=timezone.utc) @@ -1994,6 +2104,26 @@ async def _maintain_steward_placement(self) -> None: await self._teardown_steward_instances(extras) return if stewards: + # Reconcile a preference-list edit: a steward whose model was + # removed from ``steward_models`` is deliberately deselected and + # must not keep serving indefinitely. Teardown here is enough — + # this same invariant re-places from the current list on the + # next tick. Reordering the list alone never replaces a placed + # steward (upgrade churn is worse than a working older brain). + placed = self.state.instances.get(stewards[0]) + if placed is not None: + placed_model = str(placed.shard_assignments.model_id) + if placed_model not in fabric.steward_models: + logger.info( + f"Steward model {placed_model} was removed from " + "steward_models; replacing the placement" + ) + await self._teardown_steward_instances(stewards) + # This teardown is an intentional replacement: open the + # pacing window so the invariant really can re-place on + # the next tick, as promised above, instead of waiting + # out a window started by the previous placement. + self._steward_last_attempt_monotonic = time.monotonic() - 60.0 return now = time.monotonic() @@ -2011,6 +2141,12 @@ async def _maintain_steward_placement(self) -> None: f"Steward model {model_ref} has no model card; skipping" ) continue + if not steward_candidate_is_servable(card): + logger.warning( + f"Steward model {model_ref} is not a tool-calling text " + "card; skipping" + ) + continue command = PlaceInstance( model_card=card, sharding=Sharding.Pipeline, @@ -2043,6 +2179,18 @@ async def _maintain_steward_placement(self) -> None: f"Steward placement with {model_ref} not possible yet: {err}" ) continue + minted_instances = { + instance_id: instance + for instance_id, instance in final_placement.items() + if instance_id not in self.state.instances + } + if placement_may_select_parserless_vllm(card, minted_instances): + logger.warning( + f"Steward model {model_ref} resolved (or may fall back) to " + "the vllm engine without a pinned tool-call parser; " + "skipping" + ) + continue logger.info( f"Establishing steward placement with {model_ref} " "(intelligent fabric)" diff --git a/src/skulk/master/tests/test_steward_placement.py b/src/skulk/master/tests/test_steward_placement.py index 256a0bdb..443364a2 100644 --- a/src/skulk/master/tests/test_steward_placement.py +++ b/src/skulk/master/tests/test_steward_placement.py @@ -166,3 +166,181 @@ def test_mlx_ring_meta_is_benign_for_a_single_node_gguf_steward() -> None: assert isinstance(instance, MlxRingInstance) assert instance.system_role == "steward" assert len(instance.shard_assignments.node_to_runner) == 1 + + +def test_steward_candidates_require_text_and_tool_capability() -> None: + """Only tool-calling text cards may serve as steward brains (#830). + + The harness always dispatches ``TextGeneration`` with server tools, so + an operator override naming any other card shape must be skipped, not + placed. + """ + from skulk.master.main import steward_candidate_is_servable + from skulk.shared.models.model_cards import ToolingCardConfig + + def _card(tasks: list[ModelTask], *, tools: bool) -> ModelCard: + return ModelCard( + model_id=ModelId("candidate-model"), + storage_size=Memory.from_gb(3), + n_layers=12, + hidden_size=30, + supports_tensor=True, + tasks=tasks, + tooling=ToolingCardConfig(supports_tool_calling=True) if tools else None, + ) + + assert steward_candidate_is_servable( + _card([ModelTask.TextGeneration], tools=True) + ) + # A speech card cannot serve steward turns no matter what it declares. + assert not steward_candidate_is_servable( + _card([ModelTask.TextToSpeech], tools=True) + ) + # A text card without tool calling can never investigate. + assert not steward_candidate_is_servable( + _card([ModelTask.TextGeneration], tools=False) + ) + # Routing truth: bootstrap sends image/embedding/speech cards to their + # specialized runners before text-engine dispatch, so a multi-task card + # carrying any of those tasks never reaches a text runner. + assert not steward_candidate_is_servable( + _card([ModelTask.TextGeneration, ModelTask.TextToSpeech], tools=True) + ) + assert not steward_candidate_is_servable( + _card([ModelTask.TextGeneration, ModelTask.TextEmbedding], tools=True) + ) + assert not steward_candidate_is_servable( + _card([ModelTask.TextGeneration, ModelTask.TextToImage], tools=True) + ) + + +def test_vllm_only_steward_candidates_require_a_pinned_parser() -> None: + """Backend truth joins model truth in the servability gate (#833). + + A tool-declaring card whose only platform-servable engine is vllm still + fails every steward turn unless the card pins + ``runtime.vllm_tool_call_parser``: the launched server rejects + tools-bearing requests loudly, so such a candidate must be skipped. + """ + from skulk.master.main import steward_candidate_is_servable + from skulk.shared.models.model_cards import ( + RuntimeCapabilityCardConfig, + ToolingCardConfig, + ) + + def _vllm_card(parser: str | None) -> ModelCard: + return ModelCard( + model_id=ModelId("vllm-only-model"), + storage_size=Memory.from_gb(3), + n_layers=12, + hidden_size=30, + supports_tensor=True, + tasks=[ModelTask.TextGeneration], + tooling=ToolingCardConfig(supports_tool_calling=True), + runtime=( + RuntimeCapabilityCardConfig(vllm_tool_call_parser=parser) + if parser is not None + else None + ), + placement=PlacementCardConfig( + compatible_backends=frozenset({"vllm-cuda"}), + ), + ) + + assert not steward_candidate_is_servable(_vllm_card(None)) + assert steward_candidate_is_servable(_vllm_card("hermes")) + + +def test_stamped_vllm_placement_without_parser_is_rejected() -> None: + """The post-placement gate catches multi-engine cards that resolve to vllm. + + A card listing vllm alongside another engine passes the card-level + servability gate, but a fleet whose hardware makes placement stamp + vllm still needs the parser pin; the walk must skip the candidate when + the minted shards resolved to vllm without one (#833). + """ + from skulk.master.main import placement_may_select_parserless_vllm + from skulk.shared.models.model_cards import ( + RuntimeCapabilityCardConfig, + ToolingCardConfig, + ) + from skulk.shared.types.worker.instances import Instance, InstanceId + + def _card(parser: str | None) -> ModelCard: + return ModelCard( + model_id=ModelId("multi-engine-model"), + storage_size=Memory.from_gb(3), + n_layers=12, + hidden_size=30, + supports_tensor=True, + tasks=[ModelTask.TextGeneration], + tooling=ToolingCardConfig(supports_tool_calling=True), + runtime=( + RuntimeCapabilityCardConfig(vllm_tool_call_parser=parser) + if parser is not None + else None + ), + placement=PlacementCardConfig( + compatible_backends=frozenset({"vllm-cuda", "llama_server-cuda"}), + backend_preference=("vllm-cuda", "llama_server-cuda"), + ), + ) + + topology, node_memory, node_network, node_ids = fully_connected_three_nodes( + (10.0, 10.0, 10.0) + ) + + def _place(card: ModelCard) -> "dict[InstanceId, Instance]": + command = PlaceInstance( + model_card=card, + sharding=Sharding.Pipeline, + instance_meta=InstanceMeta.MlxRing, + min_nodes=1, + system_role="steward", + ) + return place_instance( + command, + topology, + {}, + node_memory, + node_network, + node_resources={ + node_id: NodeResources( + backends=frozenset({"vllm-cuda", "llama_server-cuda"}) + ) + for node_id in node_ids + }, + ) + + parserless = _place(_card(None)) + stamped = { + shard.resolved_backend + for instance in parserless.values() + for shard in instance.shard_assignments.runner_to_shard.values() + } + assert stamped == {"vllm-cuda"}, "test premise: placement resolves vllm" + assert placement_may_select_parserless_vllm(_card(None), parserless) + assert not placement_may_select_parserless_vllm( + _card("hermes"), _place(_card("hermes")) + ) + + # Telemetry warm-up: no NodeResources means placement stamps no + # backend, and the worker's local fallback could still pick vllm, so + # an unstamped parserless placement must read as unsafe too. + unstamped_command = PlaceInstance( + model_card=_card(None), + sharding=Sharding.Pipeline, + instance_meta=InstanceMeta.MlxRing, + min_nodes=1, + system_role="steward", + ) + unstamped = place_instance( + unstamped_command, topology, {}, node_memory, node_network + ) + stamped_backends = { + shard.resolved_backend + for instance in unstamped.values() + for shard in instance.shard_assignments.runner_to_shard.values() + } + assert stamped_backends == {None}, "test premise: warm-up leaves no stamp" + assert placement_may_select_parserless_vllm(_card(None), unstamped) diff --git a/src/skulk/worker/runner/vllm/runner.py b/src/skulk/worker/runner/vllm/runner.py index 4125cd38..280df28e 100644 --- a/src/skulk/worker/runner/vllm/runner.py +++ b/src/skulk/worker/runner/vllm/runner.py @@ -834,9 +834,10 @@ def _generate(self, task: Task) -> None: if task.task_params.tools and self._tool_call_parser is None: raise RuntimeError( "This model's card declares no vLLM tool-call parser " - "(runtime.vllm_tool_call_parser or a family default), so " - "the server was launched without tool support. Retry " - "without tools or serve a tool-capable card." + "(runtime.vllm_tool_call_parser; there is no family " + "fallback), so the server was launched without tool " + "support. Retry without tools or serve a card that pins " + "a parser." ) if wants_logprobs( task.task_params.logprobs, task.task_params.top_logprobs diff --git a/website/docs/api-guide.md b/website/docs/api-guide.md index de2f9622..e43100eb 100644 --- a/website/docs/api-guide.md +++ b/website/docs/api-guide.md @@ -1317,7 +1317,10 @@ Requests cancellation of one in-flight generation command by its command ID. It covers text generation, image generation, embeddings, and speech synthesis or transcription commands owned by the API node you call: Skulk closes the local response stream and sends a task cancellation so the serving runner -stops instead of generating into the void. +stops instead of generating into the void. Steward turns (`skulk/steward`) +are covered too: the ID advertised on a steward response cancels the whole +turn — the in-flight generation stops and the investigation loop ends +instead of proceeding to its next step. Finding the command ID: @@ -1334,11 +1337,12 @@ curl -X POST http://localhost:52415/v1/cancel/ ``` A cancelled command returns -`{"message": "Command cancelled.", "command_id": "..."}`. An unknown or -already-completed command returns **404 Command not found or already -completed**. Command streams are node-local, so call the same API node that -accepted the original request. Simply disconnecting from a streaming response -triggers the same cancellation path implicitly. +`{"message": "Command cancelled.", "command_id": "..."}`; a cancelled steward +turn returns `{"message": "Steward turn cancelled.", "command_id": "..."}`. +An unknown or already-completed command returns **404 Command not found or +already completed**. Command streams are node-local, so call the same API +node that accepted the original request. Simply disconnecting from a +streaming response triggers the same cancellation path implicitly. ## Model Discovery