move process lifecycle to the job (#1266)

Claude-Session: https://claude.ai/code/session_01Cd6JuxiVXYNPCDii9go2eh

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>

facefusion/apis/endpoints/jobs.py

@@ -8,7 +8,7 @@ from starlette.responses import JSONResponse
 from starlette.status import HTTP_200_OK, HTTP_201_CREATED, HTTP_202_ACCEPTED, HTTP_400_BAD_REQUEST, HTTP_404_NOT_FOUND, HTTP_409_CONFLICT
 
 import facefusion.core
-from facefusion import args_helper, state_manager, translator
+from facefusion import args_helper, process_manager, state_manager, translator
 from facefusion.apis import jobs_helper
 from facefusion.apis.job_validator import validate_status
 from facefusion.filesystem import create_directory, get_file_extension, is_directory
@@ -108,7 +108,8 @@ async def update_jobs(request : Request) -> JSONResponse:
 
 	if action == 'cancel':
 		if job_manager.cancel_jobs(state_manager.get_item('halt_on_error')):
-			jobs_helper.stop_fork_session()
+			if not process_manager.is_pending():
+				process_manager.stop()
 
 			return JSONResponse(
 			{
@@ -121,6 +122,12 @@ async def update_jobs(request : Request) -> JSONResponse:
 		}, status_code = HTTP_400_BAD_REQUEST)
 
 	if action == 'run':
+		if not process_manager.is_pending():
+			return JSONResponse(
+			{
+				'message': translator.get('job_all_not_started', 'facefusion.apis')
+			}, status_code = HTTP_409_CONFLICT)
+
 		if job_manager.find_job_ids('queued'):
 			run_jobs_task = BackgroundTask(partial(job_runner.run_jobs, facefusion.core.process_step, state_manager.get_item('halt_on_error')))
 
@@ -135,6 +142,12 @@ async def update_jobs(request : Request) -> JSONResponse:
 		}, status_code = HTTP_400_BAD_REQUEST)
 
 	if action == 'retry':
+		if not process_manager.is_pending():
+			return JSONResponse(
+			{
+				'message': translator.get('job_all_not_retried', 'facefusion.apis')
+			}, status_code = HTTP_409_CONFLICT)
+
 		if job_manager.find_job_ids('queued'):
 			return JSONResponse(
 			{
@@ -187,7 +200,7 @@ async def update_job(request : Request) -> JSONResponse:
 
 		if job_manager.cancel_job(job_id):
 			if is_started:
-				jobs_helper.stop_fork_session()
+				process_manager.stop()
 
 			return JSONResponse(
 			{
@@ -200,6 +213,12 @@ async def update_job(request : Request) -> JSONResponse:
 		}, status_code = HTTP_400_BAD_REQUEST)
 
 	if action == 'run':
+		if not process_manager.is_pending():
+			return JSONResponse(
+			{
+				'message': translator.get('job_not_started', 'facefusion.apis')
+			}, status_code = HTTP_409_CONFLICT)
+
 		if job_id in job_manager.find_job_ids('queued'):
 			run_job_tasks = BackgroundTasks()
 			run_job_tasks.add_task(partial(job_runner.run_job, job_id, facefusion.core.process_step))
@@ -216,6 +235,12 @@ async def update_job(request : Request) -> JSONResponse:
 		}, status_code = HTTP_400_BAD_REQUEST)
 
 	if action == 'retry':
+		if not process_manager.is_pending():
+			return JSONResponse(
+			{
+				'message': translator.get('job_not_retried', 'facefusion.apis')
+			}, status_code = HTTP_409_CONFLICT)
+
 		if job_manager.find_job_ids('queued'):
 			return JSONResponse(
 			{

facefusion/apis/jobs_helper.py

@@ -1,4 +1,3 @@
-from facefusion import process_manager, session_context, session_manager
 from facefusion.apis import asset_store
 from facefusion.filesystem import is_file
 from facefusion.jobs import job_manager
@@ -14,12 +13,3 @@ def create_output_asset(job_id : JobId) -> None:
 		if output_path and is_file(output_path):
 			asset_store.create_asset('output', output_path)
 
-
-def stop_fork_session() -> None:
-	owner_id = session_manager.resolve_owner_id()
-	fork_id = session_manager.find_fork_session_id(owner_id)
-
-	if fork_id:
-		session_context.set_session_id(fork_id)
-		process_manager.stop()
-		session_context.set_session_id(owner_id)

facefusion/benchmarker.py

@@ -6,7 +6,7 @@ from time import perf_counter
 from typing import Iterator, List
 
 import facefusion.choices
-from facefusion import content_analyser, core, face_store, state_manager
+from facefusion import content_analyser, core, face_store, process_manager, state_manager
 from facefusion.cli_helper import render_table
 from facefusion.download import conditional_download, resolve_download_url
 from facefusion.filesystem import get_file_extension
@@ -56,6 +56,8 @@ def cycle(cycle_count : int) -> BenchmarkCycleSet:
 	video_frame_total = count_video_frame_total(state_manager.get_item('target_path'))
 	state_manager.init_item('output_video_fps', detect_video_fps(state_manager.get_item('target_path')))
 
+	process_manager.start()
+
 	if state_manager.get_item('benchmark_mode') == 'warm':
 		core.conditional_process()
 
@@ -70,6 +72,8 @@ def cycle(cycle_count : int) -> BenchmarkCycleSet:
 		end_time = perf_counter()
 		process_times.append(end_time - start_time)
 
+	process_manager.end()
+
 	average_run = round(statistics.mean(process_times), 2)
 	fastest_run = round(min(process_times), 2)
 	slowest_run = round(max(process_times), 2)

facefusion/core.py

@@ -342,7 +342,6 @@ def process_step(job_id : JobId, step_index : StepIndex, step_args : Args) -> Er
 		[
 			state_manager.STATE_SET,
 			content_store.CONTENT_STORE,
-			process_manager.PROCESS_STORE,
 			video_manager.VIDEO_POOL_STORE
 		]
 

facefusion/download.py

@@ -74,6 +74,7 @@ def ping_static_url(url : str) -> bool:
 
 
 def conditional_download_files(file_set : DownloadSet) -> bool:
+	process_state = process_manager.get_state()
 	process_manager.check()
 
 	for file in file_set.values():
@@ -84,7 +85,9 @@ def conditional_download_files(file_set : DownloadSet) -> bool:
 			conditional_download(os.path.dirname(file_path), [ file_url ])
 
 	is_valid = conditional_validate_files(file_set)
-	process_manager.end()
+
+	if process_manager.is_checking():
+		process_manager.set_state(process_state)
 
 	return is_valid
 
@@ -118,6 +121,7 @@ def resolve_download_url(base_name : str, file_name : str) -> Optional[str]:
 
 	for download_provider in download_providers:
 		download_url = resolve_download_url_by_provider(download_provider, base_name, file_name)
+
 		if download_url:
 			return download_url
 

facefusion/ffmpeg.py

@@ -346,8 +346,7 @@ def concat_video(output_path : str, temp_output_paths : List[str]) -> bool:
 		ffmpeg_builder.set_faststart(output_video_format),
 		ffmpeg_builder.force_output(output_path)
 	)
-	process = run_ffmpeg(commands)
-	process.communicate()
+	process = await_process(run_ffmpeg(commands))
 	remove_file(concat_video_path)
 	return process.returncode == 0
 

facefusion/jobs/job_runner.py

@@ -1,5 +1,6 @@
 import os
 
+from facefusion import process_manager
 from facefusion.ffmpeg import concat_video
 from facefusion.filesystem import are_images, are_videos, copy_file, create_directory, is_directory, is_file, move_directory, move_file, remove_directory, remove_file, resolve_file_paths
 from facefusion.jobs import job_helper, job_manager
@@ -10,11 +11,16 @@ def run_job(job_id : JobId, process_step : ProcessStep) -> bool:
 	queued_job_ids = job_manager.find_job_ids('queued')
 
 	if job_id in queued_job_ids:
+		process_manager.start()
+
 		if run_steps(job_id, process_step) and job_id in job_manager.find_job_ids('queued') and finalize_steps(job_id):
 			clean_steps(job_id)
+			process_manager.end()
 			return job_manager.move_job_file(job_id, 'completed')
 
 		clean_steps(job_id)
+		process_manager.end()
+
 		if job_id in job_manager.find_job_ids('queued'):
 			job_manager.move_job_file(job_id, 'failed')
 	return False

facefusion/process_manager.py

@@ -12,14 +12,13 @@ def init() -> None:
 
 
 def get_state() -> ProcessState:
-	session_id = get_session_id()
 	owner_id = resolve_owner_id()
 	local_id = resolve_local_id()
 
-	if store_creator.get_content(PROCESS_STORE, owner_id) == 'stopping' or store_creator.get_content(PROCESS_STORE, local_id) == 'stopping':
+	if store_creator.get_content(PROCESS_STORE, local_id) == 'stopping':
 		return 'stopping'
 
-	return store_creator.get_content(PROCESS_STORE, session_id)
+	return store_creator.get_content(PROCESS_STORE, owner_id)
 
 
 def is_checking() -> bool:
@@ -39,8 +38,8 @@ def is_pending() -> bool:
 
 
 def set_state(process_state : ProcessState) -> None:
-	session_id = get_session_id()
-	store_creator.set_content(PROCESS_STORE, session_id, process_state)
+	owner_id = resolve_owner_id()
+	store_creator.set_content(PROCESS_STORE, owner_id, process_state)
 
 
 def check() -> None:

facefusion/workflows/modules/audio_to_image.py

@@ -1,6 +1,5 @@
 from functools import partial
 
-from facefusion import process_manager
 from facefusion.types import ErrorCode
 from facefusion.workflows.as_frames import create_temp_frames
 from facefusion.workflows.core import clear, process_frames, setup
@@ -22,14 +21,10 @@ def process(start_time : float) -> ErrorCode:
 		clear
 	]
 
-	process_manager.start()
-
 	for task in tasks:
 		error_code = task() #type:ignore[operator]
 
 		if error_code > 0:
-			process_manager.end()
 			return error_code
 
-	process_manager.end()
 	return 0

facefusion/workflows/modules/audio_to_image_as_frames.py

@@ -1,6 +1,5 @@
 from functools import partial
 
-from facefusion import process_manager
 from facefusion.types import ErrorCode
 from facefusion.workflows.as_frames import copy_temp_frames, create_temp_frames, finalize_frames
 from facefusion.workflows.core import clear, process_frames, setup
@@ -20,14 +19,10 @@ def process(start_time : float) -> ErrorCode:
 		clear
 	]
 
-	process_manager.start()
-
 	for task in tasks:
 		error_code = task() #type:ignore[operator]
 
 		if error_code > 0:
-			process_manager.end()
 			return error_code
 
-	process_manager.end()
 	return 0

facefusion/workflows/modules/image_to_image.py

@@ -1,6 +1,5 @@
 from functools import partial
 
-from facefusion import process_manager
 from facefusion.types import ErrorCode
 from facefusion.workflows.core import clear, setup
 from facefusion.workflows.to_image import analyse_image, finalize_image, prepare_image, process_image
@@ -18,14 +17,10 @@ def process(start_time : float) -> ErrorCode:
 		clear
 	]
 
-	process_manager.start()
-
 	for task in tasks:
 		error_code = task() #type:ignore[operator]
 
 		if error_code > 0:
-			process_manager.end()
 			return error_code
 
-	process_manager.end()
 	return 0

facefusion/workflows/modules/image_to_video.py

@@ -1,6 +1,6 @@
 from functools import partial
 
-from facefusion import process_manager, state_manager
+from facefusion import state_manager
 from facefusion.types import ErrorCode
 from facefusion.workflows.core import clear, process_frames, setup
 from facefusion.workflows.to_video import analyse_video, create_temp_frames, finalize_video, merge_frames, process_memory_frames, restore_audio
@@ -32,14 +32,10 @@ def process(start_time : float) -> ErrorCode:
 		clear
 	])
 
-	process_manager.start()
-
 	for task in tasks:
 		error_code = task() #type:ignore[operator]
 
 		if error_code > 0:
-			process_manager.end()
 			return error_code
 
-	process_manager.end()
 	return 0

facefusion/workflows/modules/image_to_video_as_frames.py

@@ -1,6 +1,5 @@
 from functools import partial
 
-from facefusion import process_manager
 from facefusion.types import ErrorCode
 from facefusion.workflows.as_frames import copy_temp_frames, finalize_frames
 from facefusion.workflows.core import clear, process_frames, setup
@@ -20,14 +19,10 @@ def process(start_time : float) -> ErrorCode:
 		clear
 	]
 
-	process_manager.start()
-
 	for task in tasks:
 		error_code = task() #type:ignore[operator]
 
 		if error_code > 0:
-			process_manager.end()
 			return error_code
 
-	process_manager.end()
 	return 0

tests/apis/endpoints/test_jobs.py

@@ -355,9 +355,8 @@ def test_cancel_jobs(test_client : TestClient) -> None:
 	{
 		'Authorization': 'Bearer ' + access_token
 	})
-	fork_id = session_manager.fork_session()
-	process_manager.start()
 	session_context.set_session_id(session_id)
+	process_manager.start()
 
 	cancel_jobs_response = test_client.patch('/jobs?action=cancel', headers =
 	{
@@ -367,11 +366,10 @@ def test_cancel_jobs(test_client : TestClient) -> None:
 
 	assert cancel_jobs_body.get('message') == 'ok'
 	assert find_job_ids('drafted') == [ 'job-test-cancel-jobs' ]
-	assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'stopping'
+	assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'stopping'
 	assert cancel_jobs_response.status_code == 200
 
-	session_manager.clear_cli_session(fork_id)
-	store_creator.delete_content(process_manager.PROCESS_STORE, fork_id)
+	process_manager.end()
 
 
 def test_cancel_job(test_client : TestClient) -> None:
@@ -410,9 +408,8 @@ def test_cancel_job(test_client : TestClient) -> None:
 	{
 		'Authorization': 'Bearer ' + access_token
 	})
-	fork_id = session_manager.fork_session()
-	process_manager.start()
 	session_context.set_session_id(session_id)
+	process_manager.start()
 
 	cancel_job_response = test_client.patch('/jobs/job-test-cancel-job?action=cancel', headers =
 	{
@@ -422,7 +419,7 @@ def test_cancel_job(test_client : TestClient) -> None:
 
 	assert cancel_job_body.get('message') == 'ok'
 	assert find_job_ids('drafted') == [ 'job-test-cancel-job' ]
-	assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'processing'
+	assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'processing'
 	assert cancel_job_response.status_code == 200
 
 	test_client.patch('/jobs/job-test-cancel-job?action=submit', headers =
@@ -439,11 +436,10 @@ def test_cancel_job(test_client : TestClient) -> None:
 
 	assert cancel_job_body.get('message') == 'ok'
 	assert find_job_ids('drafted') == [ 'job-test-cancel-job' ]
-	assert store_creator.get_content(process_manager.PROCESS_STORE, fork_id) == 'stopping'
+	assert store_creator.get_content(process_manager.PROCESS_STORE, session_id) == 'stopping'
 	assert cancel_job_response.status_code == 200
 
-	session_manager.clear_cli_session(fork_id)
-	store_creator.delete_content(process_manager.PROCESS_STORE, fork_id)
+	process_manager.end()
 
 
 def test_run_jobs(test_client : TestClient) -> None:
@@ -482,6 +478,28 @@ def test_run_jobs(test_client : TestClient) -> None:
 		'Authorization': 'Bearer ' + access_token
 	})
 
+	process_manager.start()
+
+	run_jobs_response = test_client.patch('/jobs?action=run', headers =
+	{
+		'Authorization': 'Bearer ' + access_token
+	})
+	run_jobs_body = run_jobs_response.json()
+
+	assert run_jobs_body.get('message') == 'jobs not started'
+	assert run_jobs_response.status_code == 409
+
+	process_manager.check()
+
+	run_jobs_response = test_client.patch('/jobs?action=run', headers =
+	{
+		'Authorization': 'Bearer ' + access_token
+	})
+
+	assert run_jobs_response.status_code == 409
+
+	process_manager.end()
+
 	with patch('facefusion.jobs.job_runner.run_jobs', return_value = True) as run_jobs_mock:
 		run_jobs_response = test_client.patch('/jobs?action=run', headers =
 		{
@@ -531,6 +549,19 @@ def test_run_job(test_client : TestClient) -> None:
 		'Authorization': 'Bearer ' + access_token
 	})
 
+	process_manager.start()
+
+	run_job_response = test_client.patch('/jobs/job-test-run-job?action=run', headers =
+	{
+		'Authorization': 'Bearer ' + access_token
+	})
+	run_job_body = run_job_response.json()
+
+	assert run_job_body.get('message') == 'job not started'
+	assert run_job_response.status_code == 409
+
+	process_manager.end()
+
 	with patch('facefusion.jobs.job_runner.run_job', return_value = True) as run_job_mock:
 		run_job_response = test_client.patch('/jobs/job-test-run-job?action=run', headers =
 		{
@@ -581,6 +612,19 @@ def test_retry_jobs(test_client : TestClient) -> None:
 	set_steps_status('job-test-retry-jobs', 'failed')
 	move_job_file('job-test-retry-jobs', 'failed')
 
+	process_manager.start()
+
+	retry_jobs_response = test_client.patch('/jobs?action=retry', headers =
+	{
+		'Authorization': 'Bearer ' + access_token
+	})
+	retry_jobs_body = retry_jobs_response.json()
+
+	assert retry_jobs_body.get('message') == 'jobs not retried'
+	assert retry_jobs_response.status_code == 409
+
+	process_manager.end()
+
 	with patch('facefusion.jobs.job_runner.retry_jobs', return_value = True) as retry_jobs_mock:
 		retry_jobs_response = test_client.patch('/jobs?action=retry', headers =
 		{
@@ -642,6 +686,19 @@ def test_retry_job(test_client : TestClient) -> None:
 	set_steps_status('job-test-retry-job', 'failed')
 	move_job_file('job-test-retry-job', 'failed')
 
+	process_manager.start()
+
+	retry_job_response = test_client.patch('/jobs/job-test-retry-job?action=retry', headers =
+	{
+		'Authorization': 'Bearer ' + access_token
+	})
+	retry_job_body = retry_job_response.json()
+
+	assert retry_job_body.get('message') == 'job not retried'
+	assert retry_job_response.status_code == 409
+
+	process_manager.end()
+
 	with patch('facefusion.jobs.job_runner.retry_job', return_value = True) as retry_job_mock:
 		retry_job_response = test_client.patch('/jobs/job-test-retry-job?action=retry', headers =
 		{

tests/jobs/test_job_runner.py

@@ -110,8 +110,15 @@ def test_run_job() -> None:
 	assert run_job('job-test-run-job', process_step) is False
 
 	submit_job('job-test-run-job')
+	process_manager.stop()
 
-	assert run_job('job-test-run-job', process_step) is True
+	with patch('facefusion.process_manager.start', wraps = process_manager.start) as start_mock:
+		with patch('facefusion.process_manager.end', wraps = process_manager.end) as end_mock:
+			assert run_job('job-test-run-job', process_step) is True
+
+	assert start_mock.call_count == 1
+	assert end_mock.call_count == 1
+	assert process_manager.get_state() == 'pending'
 
 	create_job('job-test-run-job-cancel')
 	add_step('job-test-run-job-cancel', args_1)
@@ -319,6 +326,7 @@ def test_finalize_steps() -> None:
 	create_directory(temp_directory_2)
 	copy_file(args_4.get('target_path'), os.path.join(temp_directory_1, '00000001.jpg'))
 	copy_file(args_5.get('target_path'), os.path.join(temp_directory_2, '00000002.jpg'))
+	process_manager.start()
 
 	assert finalize_steps('job-test-finalize-steps') is True
 	assert is_test_output_file('output-1.mp4') is True

tests/test_download.py

@@ -1,8 +1,17 @@
+from unittest.mock import patch
+
+import pytest
+
 from facefusion import process_manager
 from facefusion.download import conditional_download_files, conditional_validate_files, get_static_download_size, ping_static_url, resolve_download_url_by_provider, validate_file
 from facefusion.types import DownloadSet
 
 
+@pytest.fixture(scope = 'module', autouse = True)
+def before_all() -> None:
+	process_manager.clear()
+
+
 def test_get_static_download_size() -> None:
 	assert get_static_download_size('https://github.com/facefusion/facefusion-assets/releases/download/models-3.4.0/yunet_2023_mar.onnx') == 232475
 	assert get_static_download_size('https://huggingface.co/facefusion/models-3.4.0/resolve/main/yunet_2023_mar.onnx') == 232475
@@ -49,6 +58,18 @@ def test_conditional_download_files() -> None:
 	assert conditional_download_files(invalid_set) is False
 	assert process_manager.is_pending() is True
 
+	process_manager.start()
+
+	assert conditional_download_files(hash_set) is True
+	assert process_manager.get_state() == 'processing'
+
+	with patch('facefusion.process_manager.check', side_effect = process_manager.stop):
+		assert conditional_download_files(hash_set) is True
+
+	assert process_manager.get_state() == 'stopping'
+
+	process_manager.end()
+
 
 def test_conditional_validate_files() -> None:
 	file_set : DownloadSet =\

tests/test_process_manager.py

@@ -42,8 +42,16 @@ def test_get_state() -> None:
 	assert get_state() == 'processing'
 
 	fork_id = session_manager.fork_session()
-	clear()
+	set_state('stopping')
+
+	assert get_state() == 'stopping'
+
+	session_context.set_session_id('session-a')
+
+	assert get_state() == 'stopping'
+
 	set_state('processing')
+	session_context.set_session_id(fork_id)
 
 	assert get_state() == 'processing'