diff --git a/py/packages/genkit-google-genai/tests/google_genai_plugin_test.py b/py/packages/genkit-google-genai/tests/google_genai_plugin_test.py
index 94173872eb..12c5f5b7d7 100644
--- a/py/packages/genkit-google-genai/tests/google_genai_plugin_test.py
+++ b/py/packages/genkit-google-genai/tests/google_genai_plugin_test.py
@@ -50,7 +50,7 @@
 from genkit_google_genai.models.imagen import ImagenConfigSchema
 from genkit_google_genai.models.veo import VeoConfigSchema, VeoModel
 
-from genkit import ActionKind, GenkitError, Message, ModelRequest, Part, Role, TextPart
+from genkit import ActionKind, Genkit, GenkitError, Message, ModelRequest, Part, Role, TextPart
 from genkit.model import Operation
 from genkit.plugin_api import Action, to_json_schema
 
@@ -563,6 +563,21 @@ async def test_vertexai_resolve_veo_as_model_returns_none(mock_list_models: Magi
     assert action is None
 
 
+@patch('genkit_google_genai.google.genai.client.Client')
+@patch('genkit_google_genai.google._list_genai_models')
+@pytest.mark.asyncio
+async def test_resolve_model_finds_veo_as_background(mock_list_models: MagicMock, mock_client: MagicMock) -> None:
+    """resolve(MODEL, veo) is None so resolve_model can see the background start action."""
+    mock_list_models.return_value = GenaiModels()
+
+    ai = Genkit(plugins=[GoogleAI(api_key='test-key')])
+    action = await ai.registry.resolve_model('googleai/veo-3.0-generate-001')
+
+    assert action is not None
+    assert action.kind == ActionKind.BACKGROUND_MODEL
+    assert action.name == 'googleai/veo-3.0-generate-001'
+
+
 @patch('genkit_google_genai.google.genai.client.Client')
 @patch('genkit_google_genai.google._list_genai_models')
 @pytest.mark.asyncio
diff --git a/py/packages/genkit/src/genkit/_ai/_generate.py b/py/packages/genkit/src/genkit/_ai/_generate.py
index a2c4356612..fdfdda9420 100644
--- a/py/packages/genkit/src/genkit/_ai/_generate.py
+++ b/py/packages/genkit/src/genkit/_ai/_generate.py
@@ -72,6 +72,7 @@
     FinishReason,
     MiddlewareRef,
     MultipartToolResponse,
+    Operation,
     Part,
     Role,
     TextPart,
@@ -636,6 +637,17 @@ def handler(chunk: ModelResponseChunk[Any]) -> None:
             ctx.replace_on_chunk(previous)
 
 
+def _latency_ms_from_operation(operation: Operation) -> float | None:
+    """Copy start timing off the handle if the wrapper stamped it."""
+    meta = operation.metadata
+    if not isinstance(meta, dict):
+        return None
+    raw_ms = meta.get('latencyMs')
+    if isinstance(raw_ms, int | float):
+        return float(raw_ms)
+    return None
+
+
 def _persist_threaded_conversation(response: ModelResponse, messages: list[Message]) -> ModelResponse:
     """Persist the threaded conversation onto the response's request.
 
@@ -663,6 +675,15 @@ async def _generate_action_turn(
 
     model, tools, format_def = await resolve_parameters(registry, raw_request)
 
+    if model.kind == ActionKind.BACKGROUND_MODEL and raw_request.resume is not None:
+        raise GenkitError(
+            status='FAILED_PRECONDITION',
+            message=(
+                f"Cannot resume background model '{model.name}'; "
+                'a background start cannot satisfy an interrupted tool turn'
+            ),
+        )
+
     raw_request, formatter = apply_format(raw_request, format_def)
 
     if raw_request.resources:
@@ -766,7 +787,7 @@ async def run_one_iteration(
             request = _augment_with_context(request)
 
         async def next_fn(params: ModelHookParams, c: GenerateMiddlewareContext) -> ModelResponse:
-            return (
+            raw = (
                 await model.run(
                     input=params.request,
                     context=c.custom_context,
@@ -774,6 +795,37 @@ async def next_fn(params: ModelHookParams, c: GenerateMiddlewareContext) -> Mode
                     abort_signal=c.abort_signal,
                 )
             ).response
+            # wrap_model reads .message. A background start is a poll handle,
+            # so box it before the hook. A chat model that returns a handle
+            # is registered on the wrong kind.
+            if model.kind == ActionKind.BACKGROUND_MODEL:
+                if isinstance(raw, ModelResponse):
+                    if raw.operation is None:
+                        raise GenkitError(
+                            status='FAILED_PRECONDITION',
+                            message=f"Background model '{model.name}' did not return an operation",
+                        )
+                    if raw.latency_ms is None:
+                        raw.latency_ms = _latency_ms_from_operation(raw.operation)
+                    return raw
+                if not isinstance(raw, Operation):
+                    raise GenkitError(
+                        status='FAILED_PRECONDITION',
+                        message=f"Background model '{model.name}' did not return an operation",
+                    )
+                return ModelResponse(
+                    operation=raw,
+                    request=params.request,
+                    latency_ms=_latency_ms_from_operation(raw),
+                )
+            if isinstance(raw, Operation) or (isinstance(raw, ModelResponse) and raw.operation is not None):
+                raise GenkitError(
+                    status='FAILED_PRECONDITION',
+                    message=(
+                        f"Model '{model.name}' is a define_model and returned an operation; use define_background_model"
+                    ),
+                )
+            return raw
 
         with chunks.intercept_model_stream(ctx, role=Role.MODEL):
             model_response = await dispatch_model(
diff --git a/py/packages/genkit/src/genkit/_core/_model.py b/py/packages/genkit/src/genkit/_core/_model.py
index 077a5412a0..66f82d9b55 100644
--- a/py/packages/genkit/src/genkit/_core/_model.py
+++ b/py/packages/genkit/src/genkit/_core/_model.py
@@ -517,9 +517,15 @@ def assert_valid_schema(self) -> None:
         pass
 
     def __eq__(self, other: object) -> bool:
-        """Compare responses by message and finish_reason."""
+        """Compare responses by message, finish_reason, and operation id.
+
+        Two start handles with different job ids are not the same response.
+        Timing on the handle is not part of the job.
+        """
         if isinstance(other, ModelResponse):
-            return self.message == other.message and self.finish_reason == other.finish_reason
+            self_op = self.operation.id if self.operation is not None else None
+            other_op = other.operation.id if other.operation is not None else None
+            return self.message == other.message and self.finish_reason == other.finish_reason and self_op == other_op
         return super().__eq__(other)
 
     def __hash__(self) -> int:
diff --git a/py/packages/genkit/tests/genkit/ai/background_generate_test.py b/py/packages/genkit/tests/genkit/ai/background_generate_test.py
new file mode 100644
index 0000000000..c610a8f980
--- /dev/null
+++ b/py/packages/genkit/tests/genkit/ai/background_generate_test.py
@@ -0,0 +1,365 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"""generate() / generate_operation() against a define_background_model fake."""
+
+from collections.abc import Awaitable, Callable
+
+import pytest
+
+from genkit import ActionKind, Document, Genkit, Message
+from genkit._core._action import ActionRunContext
+from genkit._core._error import GenkitError
+from genkit._core._middleware import BaseMiddleware, GenerateHookParams, GenerateMiddlewareContext, ModelHookParams
+from genkit._core._model import ModelRequest, ModelResponse
+from genkit._core._typing import (
+    FinishReason,
+    Operation,
+    Part,
+    Role,
+    TextPart,
+    ToolRequest,
+    ToolRequestPart,
+    ToolResponse,
+    ToolResponsePart,
+)
+
+
+@pytest.fixture
+def ai() -> Genkit:
+    return Genkit()
+
+
+def register_bg_model(ai: Genkit, *, op_id: str = 'bg-op-123') -> None:
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id=op_id, done=False)
+
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(
+        name='bg-model',
+        start=start,
+        check=check,
+    )
+
+
+@pytest.mark.asyncio
+async def test_generate_returns_operation_for_background_model(ai: Genkit) -> None:
+    """generate() wraps the start handle. message stays empty."""
+    register_bg_model(ai)
+
+    response = await ai.generate(model='bg-model', prompt='a cat surfing')
+
+    assert response.operation is not None
+    assert response.operation.id == 'bg-op-123'
+    assert response.operation.done is False
+    assert response.operation.action == '/background-model/bg-model'
+    assert response.message is None
+
+
+@pytest.mark.asyncio
+async def test_generate_operation_with_background_model(ai: Genkit) -> None:
+    """generate_operation() returns that same handle."""
+    register_bg_model(ai, op_id='bg-op-456')
+
+    operation = await ai.generate_operation(model='bg-model', prompt='a cat surfing')
+
+    assert isinstance(operation, Operation)
+    assert operation.id == 'bg-op-456'
+    assert operation.action == '/background-model/bg-model'
+
+
+@pytest.mark.asyncio
+async def test_generate_returns_the_job_without_polling(ai: Genkit) -> None:
+    """generate() hands back the job now. It does not wait until the job is done.
+
+    A background model (video, and anything registered with
+    ``define_background_model``) starts a job and returns a handle. You
+    poll later with ``check_operation``. ``generate()`` and
+    ``generate_operation()`` only start; they must not call ``check``
+    on the way out, or a long render would block the first call.
+    """
+    checks = 0
+
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id='bg-op-123', done=False)
+
+    async def check(op: Operation) -> Operation:
+        nonlocal checks
+        checks += 1
+        return Operation(id=op.id, done=True)
+
+    ai.define_background_model(name='bg-model', start=start, check=check)
+
+    response = await ai.generate(model='bg-model', prompt='a cat surfing')
+    operation = await ai.generate_operation(model='bg-model', prompt='a cat surfing')
+
+    assert response.operation is not None
+    assert response.operation.done is False
+    assert operation.done is False
+    assert checks == 0
+
+
+class ReadsMessage(BaseMiddleware):
+    async def wrap_model(
+        self,
+        params: ModelHookParams,
+        ctx: GenerateMiddlewareContext,
+        next_fn: Callable[[ModelHookParams, GenerateMiddlewareContext], Awaitable[ModelResponse]],
+    ) -> ModelResponse:
+        response = await next_fn(params, ctx)
+        _ = response.message
+        return response
+
+
+@pytest.mark.asyncio
+async def test_generate_boxes_before_wrap_model(ai: Genkit) -> None:
+    """wrap_model sees a ModelResponse, so reading .message does not crash."""
+    register_bg_model(ai)
+
+    response = await ai.generate(model='bg-model', prompt='a cat surfing', use=[ReadsMessage()])
+
+    assert response.operation is not None
+    assert response.operation.id == 'bg-op-123'
+    assert response.message is None
+
+
+class DropsOperation(BaseMiddleware):
+    async def wrap_model(
+        self,
+        params: ModelHookParams,
+        ctx: GenerateMiddlewareContext,
+        next_fn: Callable[[ModelHookParams, GenerateMiddlewareContext], Awaitable[ModelResponse]],
+    ) -> ModelResponse:
+        resp = await next_fn(params, ctx)
+        return ModelResponse(message=resp.message, finish_reason=resp.finish_reason)
+
+
+@pytest.mark.asyncio
+async def test_generate_operation_fails_when_wrap_model_drops_operation(ai: Genkit) -> None:
+    """generate() stays quiet; generate_operation is the one missing-handle error."""
+    register_bg_model(ai)
+
+    response = await ai.generate(model='bg-model', prompt='a cat surfing', use=[DropsOperation()])
+    assert response.operation is None
+
+    with pytest.raises(GenkitError, match='did not return an operation') as exc_info:
+        await ai.generate_operation(model='bg-model', prompt='a cat surfing', use=[DropsOperation()])
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+
+
+class DropsGenerate(BaseMiddleware):
+    async def wrap_generate(
+        self,
+        params: GenerateHookParams,
+        ctx: GenerateMiddlewareContext,
+        next_fn: Callable[[GenerateHookParams, GenerateMiddlewareContext], Awaitable[ModelResponse]],
+    ) -> ModelResponse:
+        resp = await next_fn(params, ctx)
+        return ModelResponse(message=resp.message, finish_reason=resp.finish_reason)
+
+
+@pytest.mark.asyncio
+async def test_generate_operation_fails_when_wrap_generate_drops_operation(ai: Genkit) -> None:
+    """Same one error if wrap_generate rebuilds the response without the handle."""
+    register_bg_model(ai)
+
+    response = await ai.generate(model='bg-model', prompt='a cat surfing', use=[DropsGenerate()])
+    assert response.operation is None
+
+    with pytest.raises(GenkitError, match='did not return an operation') as exc_info:
+        await ai.generate_operation(model='bg-model', prompt='a cat surfing', use=[DropsGenerate()])
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+
+
+class SwallowsStart(BaseMiddleware):
+    async def wrap_model(
+        self,
+        params: ModelHookParams,
+        ctx: GenerateMiddlewareContext,
+        next_fn: Callable[[ModelHookParams, GenerateMiddlewareContext], Awaitable[ModelResponse]],
+    ) -> ModelResponse:
+        try:
+            return await next_fn(params, ctx)
+        except GenkitError:
+            return ModelResponse(
+                message=Message(role=Role.MODEL, content=[Part(root=TextPart(text='FLASH'))]),
+                finish_reason=FinishReason.STOP,
+            )
+
+
+@pytest.mark.asyncio
+async def test_generate_keeps_fallback_answer_when_start_raises(ai: Genkit) -> None:
+    """A hook that substitutes another model's answer is not a missing handle."""
+
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        raise GenkitError(status='UNAVAILABLE', message='veo capacity exhausted')
+
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='bg-model', start=start, check=check)
+
+    response = await ai.generate(model='bg-model', prompt='a cat', use=[SwallowsStart()])
+
+    assert response.text == 'FLASH'
+    assert response.operation is None
+
+
+@pytest.mark.asyncio
+async def test_generate_persists_clean_history_without_injected_docs(ai: Genkit) -> None:
+    """Injected RAG text stays off response.request.messages."""
+    register_bg_model(ai)
+
+    response = await ai.generate(
+        model='bg-model',
+        prompt='render a cat',
+        docs=[Document.from_text('SECRET-CONTEXT-DOC')],
+    )
+
+    assert response.request is not None
+    dumped = ' '.join(m.text for m in response.request.messages)
+    assert 'SECRET-CONTEXT-DOC' not in dumped
+    assert 'render a cat' in dumped
+
+
+@pytest.mark.asyncio
+async def test_generate_rejects_resume_on_background_model(ai: Genkit) -> None:
+    """A video start cannot satisfy an interrupt resume. Don't bill start()."""
+    started = 0
+
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        nonlocal started
+        started += 1
+        return Operation(id='bg-op-123', done=False)
+
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='bg-model', start=start, check=check)
+
+    with pytest.raises(GenkitError, match='Cannot resume background model') as exc_info:
+        await ai.generate(
+            model='bg-model',
+            messages=[
+                Message(role=Role.USER, content=[Part(root=TextPart(text='hi'))]),
+                Message(
+                    role=Role.MODEL,
+                    content=[
+                        Part(root=ToolRequestPart(tool_request=ToolRequest(name='ping', input={}, ref='1'))),
+                    ],
+                ),
+            ],
+            resume_respond=[ToolResponsePart(tool_response=ToolResponse(name='ping', ref='1', output='ok'))],
+        )
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+    assert started == 0
+
+
+@pytest.mark.asyncio
+async def test_generate_forwards_start_latency(ai: Genkit) -> None:
+    """wrapped_start already measured the call; the boxed response should carry it."""
+    register_bg_model(ai)
+
+    response = await ai.generate(model='bg-model', prompt='a cat')
+
+    assert response.latency_ms is not None
+    assert response.latency_ms >= 0
+
+
+def _register_raw_background(ai: Genkit, *, name: str, start: Callable[..., Awaitable[object]]) -> None:
+    ai.registry.register_action(name=name, kind=ActionKind.BACKGROUND_MODEL, fn=start)
+
+
+@pytest.mark.asyncio
+async def test_background_start_already_boxed_model_response(ai: Genkit) -> None:
+    """start() that already returned ModelResponse(operation=...) is passed through."""
+
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> ModelResponse:
+        return ModelResponse(operation=Operation(id='boxed-1', done=False))
+
+    _register_raw_background(ai, name='boxed-bg', start=start)
+
+    response = await ai.generate(model='boxed-bg', prompt='a cat')
+
+    assert response.operation is not None
+    assert response.operation.id == 'boxed-1'
+    assert response.message is None
+
+
+@pytest.mark.asyncio
+async def test_background_start_model_response_without_operation_raises(ai: Genkit) -> None:
+    """A background start that returns a chat turn has no handle to poll."""
+
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> ModelResponse:
+        return ModelResponse(
+            message=Message(role=Role.MODEL, content=[Part(root=TextPart(text='nope'))]),
+        )
+
+    _register_raw_background(ai, name='empty-bg', start=start)
+
+    with pytest.raises(GenkitError, match='Background model') as exc_info:
+        await ai.generate(model='empty-bg', prompt='a cat')
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+    assert 'did not return an operation' in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_define_model_returning_operation_raises(ai: Genkit) -> None:
+    """A chat model that returns a bare Operation is registered on the wrong kind."""
+
+    async def model_fn(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id='sneaky', done=False)
+
+    ai.define_model(name='plain', fn=model_fn)
+
+    with pytest.raises(GenkitError, match='define_background_model') as exc_info:
+        await ai.generate(model='plain', prompt='hi')
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+    assert 'plain' in str(exc_info.value)
+
+
+@pytest.mark.asyncio
+async def test_define_model_returning_model_response_with_operation_raises(ai: Genkit) -> None:
+    """A chat model that stuffs a handle onto ModelResponse is the same mistake."""
+
+    async def model_fn(_request: ModelRequest, _ctx: ActionRunContext) -> ModelResponse:
+        return ModelResponse(
+            message=Message(role=Role.MODEL, content=[Part(root=TextPart(text='Started'))]),
+            operation=Operation(id='lro-1', done=False),
+        )
+
+    ai.define_model(name='plain', fn=model_fn)
+
+    with pytest.raises(GenkitError, match='define_background_model') as exc_info:
+        await ai.generate(model='plain', prompt='hi')
+
+    assert exc_info.value.status == 'FAILED_PRECONDITION'
+
+
+def test_model_response_eq_uses_operation_id() -> None:
+    """Same job id is the same response even when start timing differs."""
+    a = ModelResponse(operation=Operation(id='unique-a', metadata={'latencyMs': 0.166}))
+    b = ModelResponse(operation=Operation(id='unique-a', metadata={'latencyMs': 0.002}))
+    c = ModelResponse(operation=Operation(id='unique-b'))
+    assert a == b
+    assert a != c
diff --git a/py/packages/genkit/tests/genkit/ai/generate_operation_test.py b/py/packages/genkit/tests/genkit/ai/generate_operation_test.py
index d3cdd75f5e..df0d2f0510 100644
--- a/py/packages/genkit/tests/genkit/ai/generate_operation_test.py
+++ b/py/packages/genkit/tests/genkit/ai/generate_operation_test.py
@@ -176,32 +176,15 @@ async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelRespons
 
 @pytest.mark.asyncio
 async def test_generate_operation_success_with_lro_model(ai: Genkit) -> None:
-    """Test successful generate_operation with a proper long-running model."""
-    expected_operation = Operation(
-        id='test-operation-123',
-        done=False,
-        action='/background-model/lro-model',
-    )
+    """Test successful generate_operation with a background model."""
 
-    # Define a model that supports long_running and returns an operation
-    async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelResponse:
-        return ModelResponse(
-            message=Message(
-                role=Role.MODEL,
-                content=[Part(root=TextPart(text='Started'))],
-            ),
-            operation=expected_operation,
-        )
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id='test-operation-123', done=False)
 
-    ai.define_model(
-        name='lro-model',
-        fn=model_fn,
-        info=ModelInfo(
-            supports=Supports(
-                long_running=True,
-            ),
-        ),
-    )
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='lro-model', start=start, check=check)
 
     operation = await ai.generate_operation(model='lro-model', prompt='Generate video')
 
@@ -212,44 +195,17 @@ async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelRespons
 
 
 @pytest.mark.asyncio
-async def test_generate_operation_with_default_model(ai: Genkit) -> None:
+async def test_generate_operation_with_default_model() -> None:
     """Test generate_operation uses default model when set."""
-    expected_operation = Operation(
-        id='default-op-456',
-        done=False,
-    )
 
-    async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelResponse:
-        return ModelResponse(
-            message=Message(
-                role=Role.MODEL,
-                content=[Part(root=TextPart(text='Started'))],
-            ),
-            operation=expected_operation,
-        )
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id='default-op-456', done=False)
 
-    ai.define_model(
-        name='default-lro-model',
-        fn=model_fn,
-        info=ModelInfo(
-            supports=Supports(
-                long_running=True,
-            ),
-        ),
-    )
+    async def check(op: Operation) -> Operation:
+        return op
 
-    # Create a new Genkit instance with the default model set
     ai_with_default = Genkit(model='default-lro-model')
-    # Re-register the model on the new instance
-    ai_with_default.define_model(
-        name='default-lro-model',
-        fn=model_fn,
-        info=ModelInfo(
-            supports=Supports(
-                long_running=True,
-            ),
-        ),
-    )
+    ai_with_default.define_background_model(name='default-lro-model', start=start, check=check)
 
     operation = await ai_with_default.generate_operation(prompt='Generate video')
 
@@ -261,28 +217,16 @@ async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelRespons
 async def test_generate_operation_passes_all_options(ai: Genkit) -> None:
     """Test that generate_operation passes all options to generate()."""
     captured_request: ModelRequest | None = None
-    expected_operation = Operation(id='opt-test-789', done=False)
 
-    async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelResponse:
+    async def start(request: ModelRequest, _ctx: ActionRunContext) -> Operation:
         nonlocal captured_request
         captured_request = request
-        return ModelResponse(
-            message=Message(
-                role=Role.MODEL,
-                content=[Part(root=TextPart(text='Started'))],
-            ),
-            operation=expected_operation,
-        )
+        return Operation(id='opt-test-789', done=False)
 
-    ai.define_model(
-        name='options-test-model',
-        fn=model_fn,
-        info=ModelInfo(
-            supports=Supports(
-                long_running=True,
-            ),
-        ),
-    )
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='options-test-model', start=start, check=check)
 
     await ai.generate_operation(
         model='options-test-model',
@@ -292,5 +236,4 @@ async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelRespons
     )
 
     assert captured_request is not None
-    # Verify config was passed
     assert captured_request.config is not None
diff --git a/py/packages/genkit/tests/genkit/ai/model_ref_generate_test.py b/py/packages/genkit/tests/genkit/ai/model_ref_generate_test.py
index 588f441d30..308159fa00 100644
--- a/py/packages/genkit/tests/genkit/ai/model_ref_generate_test.py
+++ b/py/packages/genkit/tests/genkit/ai/model_ref_generate_test.py
@@ -17,8 +17,8 @@
 from genkit._ai._testing import EchoModel, define_echo_model
 from genkit._core._action import ActionRunContext
 from genkit._core._error import GenkitError
-from genkit._core._model import Message, ModelRequest, ModelResponse
-from genkit._core._typing import ModelInfo, Operation, Part, Role, Supports, TextPart
+from genkit._core._model import ModelRequest
+from genkit._core._typing import Operation
 from genkit.model import model_ref
 
 
@@ -214,28 +214,16 @@ async def test_define_prompt_dict_none_clear_and_extra(
 @pytest.mark.asyncio
 async def test_generate_operation_with_model_ref(ai: Genkit) -> None:
     """generate_operation applies the ref's version and config, not just the name."""
-    expected_operation = Operation(
-        id='ref-op-123',
-        done=False,
-        action='/background-model/lro-model',
-    )
     seen: list[ModelRequest] = []
 
-    async def model_fn(request: ModelRequest, ctx: ActionRunContext) -> ModelResponse:
+    async def start(request: ModelRequest, _ctx: ActionRunContext) -> Operation:
         seen.append(request)
-        return ModelResponse(
-            message=Message(
-                role=Role.MODEL,
-                content=[Part(root=TextPart(text='Started'))],
-            ),
-            operation=expected_operation,
-        )
-
-    ai.define_model(
-        name='lro-model',
-        fn=model_fn,
-        info=ModelInfo(supports=Supports(long_running=True)),
-    )
+        return Operation(id='ref-op-123', done=False)
+
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='lro-model', start=start, check=check)
     ref = model_ref(
         'lro-model',
         config_schema=ModelConfig,
diff --git a/py/packages/genkit/tests/genkit/veneer/veneer_test.py b/py/packages/genkit/tests/genkit/veneer/veneer_test.py
index 77286ed08b..88f49ec299 100644
--- a/py/packages/genkit/tests/genkit/veneer/veneer_test.py
+++ b/py/packages/genkit/tests/genkit/veneer/veneer_test.py
@@ -1831,20 +1831,16 @@ async def check_fn(op: Operation) -> Operation:
 async def test_generate_operation_with_model_info_long_running(
     setup_test: SetupFixture,
 ) -> None:
-    """Verify generate_operation succeeds for a model defined with ModelInfo(supports=Supports(long_running=True))."""
+    """Verify generate_operation succeeds for a define_background_model."""
     ai, _, _, *_ = setup_test
 
-    async def my_model(request: ModelRequest) -> ModelResponse:
-        return ModelResponse(
-            message=Message(role='model', content=[TextPart(text='done')]),
-            operation=Operation(id='op123', done=False),
-        )
+    async def start(_request: ModelRequest, _ctx: ActionRunContext) -> Operation:
+        return Operation(id='op123', done=False)
 
-    ai.define_model(
-        name='lr_model',
-        fn=my_model,
-        info=ModelInfo(supports=Supports(long_running=True)),
-    )
+    async def check(op: Operation) -> Operation:
+        return op
+
+    ai.define_background_model(name='lr_model', start=start, check=check)
 
     op = await ai.generate_operation(model='lr_model', prompt='test')
     assert op is not None