diff --git a/backend/app/services/generation_jobs.py b/backend/app/services/generation_jobs.py
index 2555492..9b7c53c 100644
--- a/backend/app/services/generation_jobs.py
+++ b/backend/app/services/generation_jobs.py
@@ -21,7 +21,7 @@ def _is_terminal_status(status: str) -> bool:
 
 
 def _job_supports_queue_control(job: GenerationJob) -> bool:
-    return job.output_mode in {"story", "storybook"}
+    return job.output_mode in {"story", "storybook", "asset_generation"}
 
 
 def generation_job_can_cancel(job: GenerationJob) -> bool:
diff --git a/backend/tests/test_generation_jobs.py b/backend/tests/test_generation_jobs.py
index 3c68e36..8d2631a 100644
--- a/backend/tests/test_generation_jobs.py
+++ b/backend/tests/test_generation_jobs.py
@@ -243,6 +243,8 @@ async def test_queue_story_asset_generation_dispatches_background_job(
     assert summary["input_type"] == "image"
     assert summary["status"] == "running"
     assert summary["current_step"] == "request_accepted"
+    assert summary["can_cancel"] is True
+    assert summary["can_retry"] is False
     mock_delay.assert_called_once_with(summary["id"])
 
     job = (
@@ -305,6 +307,92 @@ async def test_asset_generation_job_worker_completes_cover_image(
     ]
 
 
+async def test_cancel_queued_asset_generation_job_marks_it_canceled(
+    db_session,
+    auth_token,
+    degraded_story_with_text,
+):
+    async def override_get_db():
+        yield db_session
+
+    app.dependency_overrides[get_db] = override_get_db
+    transport = ASGITransport(app=app)
+
+    job = await create_generation_job(
+        db_session,
+        user_id=degraded_story_with_text.user_id,
+        output_mode="asset_generation",
+        input_type="image",
+        request_payload={"story_id": degraded_story_with_text.id, "assets": ["image"]},
+        story_id=degraded_story_with_text.id,
+    )
+
+    try:
+        async with AsyncClient(transport=transport, base_url="http://test") as client:
+            client.cookies.set("access_token", auth_token)
+
+            response = await client.post(f"/api/generations/jobs/{job.id}/cancel")
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "canceled"
+        assert data["current_step"] == "generation_canceled"
+        assert data["can_cancel"] is False
+        assert data["can_retry"] is True
+    finally:
+        app.dependency_overrides.clear()
+
+
+async def test_retry_failed_asset_generation_job_requeues_new_worker_job(
+    db_session,
+    auth_token,
+    degraded_story_with_text,
+):
+    async def override_get_db():
+        yield db_session
+
+    app.dependency_overrides[get_db] = override_get_db
+    transport = ASGITransport(app=app)
+    task_delay_path = "app.tasks.generation_workflow.run_generation_workflow_task.delay"
+
+    failed_job = await create_generation_job(
+        db_session,
+        user_id=degraded_story_with_text.user_id,
+        output_mode="asset_generation",
+        input_type="image",
+        request_payload={"story_id": degraded_story_with_text.id, "assets": ["image"]},
+        story_id=degraded_story_with_text.id,
+    )
+    await finish_generation_job(
+        db_session,
+        job=failed_job,
+        story=degraded_story_with_text,
+        status="failed",
+        current_step="asset_generation_failed",
+        error_message="cover timeout",
+        message="Cover image generation failed.",
+    )
+
+    try:
+        with patch(task_delay_path) as mock_delay:
+            async with AsyncClient(transport=transport, base_url="http://test") as client:
+                client.cookies.set("access_token", auth_token)
+
+                response = await client.post(f"/api/generations/jobs/{failed_job.id}/retry")
+
+            assert response.status_code == 200
+            data = response.json()
+            assert data["id"] != failed_job.id
+            assert data["output_mode"] == "asset_generation"
+            assert data["status"] == "running"
+            assert data["current_step"] == "retry_queued"
+            assert data["can_cancel"] is True
+            assert data["can_retry"] is False
+            mock_delay.assert_called_once_with(data["id"])
+    finally:
+        app.dependency_overrides.clear()
+
+
 async def test_storybook_generation_is_queued_then_worker_records_page_image_events(
     db_session,
     auth_token,
diff --git a/docs/technical/voice-co-creation-phase-a-tech-spec.md b/docs/technical/voice-co-creation-phase-a-tech-spec.md
index 24e0834..bbb7421 100644
--- a/docs/technical/voice-co-creation-phase-a-tech-spec.md
+++ b/docs/technical/voice-co-creation-phase-a-tech-spec.md
@@ -30,6 +30,7 @@
 - finalize 会生成更稳定的标题/摘要，并在条件允许时自动排队封面补全 job
 - 已新增 `voice session analytics` 聚合指标，可跟踪 turn 成功率、ASR/TTS 失败、低置信度触发和 finalize 转化率
 - `voice session finalize` 现在会返回可追踪的 `generation_job_id`，让正式 Story 资产补全重新接回现有 generation trace 主干
+- 语音共创触发的 `asset_generation` job 现在也支持沿用统一 generation job 的取消 / 重试控制
 
 Phase A 的核心目标不是做“完全实时的语音陪伴”，而是验证以下最小闭环：