Merge pull request #1135 from trheyi/main

Refactor job handling and enhance document processing features

job/job.go

@@ -23,6 +23,21 @@ func Once(mode ModeType, data map[string]interface{}) (*Job, error) {
 	return makeJob(raw)
 }
 
+// OnceAndSave create a new job and save it immediately
+func OnceAndSave(mode ModeType, data map[string]interface{}) (*Job, error) {
+	job, err := Once(mode, data)
+	if err != nil {
+		return nil, err
+	}
+
+	err = SaveJob(job)
+	if err != nil {
+		return nil, fmt.Errorf("failed to save job: %w", err)
+	}
+
+	return job, nil
+}
+
 // Cron create a new job
 func Cron(mode ModeType, data map[string]interface{}, expression string) (*Job, error) {
 	data["mode"] = mode
@@ -35,6 +50,21 @@ func Cron(mode ModeType, data map[string]interface{}, expression string) (*Job,
 	return makeJob(raw)
 }
 
+// CronAndSave create a new cron job and save it immediately
+func CronAndSave(mode ModeType, data map[string]interface{}, expression string) (*Job, error) {
+	job, err := Cron(mode, data, expression)
+	if err != nil {
+		return nil, err
+	}
+
+	err = SaveJob(job)
+	if err != nil {
+		return nil, fmt.Errorf("failed to save job: %w", err)
+	}
+
+	return job, nil
+}
+
 // Daemon create a new job
 func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) {
 	data["mode"] = mode
@@ -46,8 +76,23 @@ func Daemon(mode ModeType, data map[string]interface{}) (*Job, error) {
 	return makeJob(raw)
 }
 
-// Start start the job
+// DaemonAndSave create a new daemon job and save it immediately
-func (j *Job) Start() error {
+func DaemonAndSave(mode ModeType, data map[string]interface{}) (*Job, error) {
+	job, err := Daemon(mode, data)
+	if err != nil {
+		return nil, err
+	}
+
+	err = SaveJob(job)
+	if err != nil {
+		return nil, fmt.Errorf("failed to save job: %w", err)
+	}
+
+	return job, nil
+}
+
+// Push pushes the job to execution queue (renamed from Start for better semantics)
+func (j *Job) Push() error {
 	// Get executions for this job
 	executions, err := j.GetExecutions()
 	if err != nil {

job/job_test.go

@@ -164,7 +164,7 @@ func TestOnceGoroutine(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = testJob.Start()
+	err = testJob.Push()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -199,7 +199,7 @@ func TestOnceProcess(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = testJob.Start()
+	err = testJob.Push()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -361,7 +361,7 @@ func TestCommand(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	err = testJob.Start()
+	err = testJob.Push()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -486,7 +486,7 @@ func TestJobExecution(t *testing.T) {
 	}
 
 	// Start the job
-	err = testJob.Start()
+	err = testJob.Push()
 	if err != nil {
 		t.Fatalf("Failed to start job: %v", err)
 	}
@@ -543,3 +543,138 @@ func TestJobExecution(t *testing.T) {
 		t.Log("No log items found, this may be expected if logging is async")
 	}
 }
+
+// TestOnceAndSave test OnceAndSave method
+func TestOnceAndSave(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerTestProcesses()
+
+	// Test OnceAndSave - should create and save job in one step
+	testJob, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Test OnceAndSave Job",
+		"description": "Job created and saved with OnceAndSave method",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create and save job: %v", err)
+	}
+
+	// Job should have a valid JobID after OnceAndSave
+	if testJob.JobID == "" {
+		t.Error("Expected job to have JobID after OnceAndSave")
+	}
+
+	// Verify job was saved to database
+	retrievedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to retrieve saved job: %v", err)
+	}
+
+	if retrievedJob.Name != "Test OnceAndSave Job" {
+		t.Errorf("Expected job name 'Test OnceAndSave Job', got '%s'", retrievedJob.Name)
+	}
+
+	// Add execution and push
+	err = testJob.Add(&job.ExecutionOptions{
+		Priority: 1,
+		SharedData: map[string]interface{}{
+			"test_data": "OnceAndSave test",
+		},
+	}, "test.job.echo", "OnceAndSave test")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	err = testJob.Push()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Give some time for execution
+	time.Sleep(2 * time.Second)
+
+	t.Log("OnceAndSave job completed successfully")
+}
+
+// TestCronAndSave test CronAndSave method
+func TestCronAndSave(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerTestProcesses()
+
+	// Test CronAndSave - should create and save cron job in one step
+	testJob, err := job.CronAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Test CronAndSave Job",
+		"description": "Cron job created and saved with CronAndSave method",
+	}, "0 0 * * *")
+	if err != nil {
+		t.Fatalf("Failed to create and save cron job: %v", err)
+	}
+
+	// Job should have a valid JobID after CronAndSave
+	if testJob.JobID == "" {
+		t.Error("Expected cron job to have JobID after CronAndSave")
+	}
+
+	// Verify cron job was saved to database
+	retrievedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to retrieve saved cron job: %v", err)
+	}
+
+	if retrievedJob.ScheduleType != string(job.ScheduleTypeCron) {
+		t.Errorf("Expected schedule type cron, got %s", retrievedJob.ScheduleType)
+	}
+
+	if retrievedJob.Name != "Test CronAndSave Job" {
+		t.Errorf("Expected job name 'Test CronAndSave Job', got '%s'", retrievedJob.Name)
+	}
+
+	t.Log("CronAndSave job created and saved successfully")
+}
+
+// TestDaemonAndSave test DaemonAndSave method
+func TestDaemonAndSave(t *testing.T) {
+	// Setup
+	test.Prepare(&testing.T{}, config.Conf)
+	defer test.Clean()
+
+	// Register test processes
+	registerTestProcesses()
+
+	// Test DaemonAndSave - should create and save daemon job in one step
+	testJob, err := job.DaemonAndSave(job.GOROUTINE, map[string]interface{}{
+		"name":        "Test DaemonAndSave Job",
+		"description": "Daemon job created and saved with DaemonAndSave method",
+	})
+	if err != nil {
+		t.Fatalf("Failed to create and save daemon job: %v", err)
+	}
+
+	// Job should have a valid JobID after DaemonAndSave
+	if testJob.JobID == "" {
+		t.Error("Expected daemon job to have JobID after DaemonAndSave")
+	}
+
+	// Verify daemon job was saved to database
+	retrievedJob, err := job.GetJob(testJob.JobID)
+	if err != nil {
+		t.Fatalf("Failed to retrieve saved daemon job: %v", err)
+	}
+
+	if retrievedJob.ScheduleType != string(job.ScheduleTypeDaemon) {
+		t.Errorf("Expected schedule type daemon, got %s", retrievedJob.ScheduleType)
+	}
+
+	if retrievedJob.Name != "Test DaemonAndSave Job" {
+		t.Errorf("Expected job name 'Test DaemonAndSave Job', got '%s'", retrievedJob.Name)
+	}
+
+	t.Log("DaemonAndSave job created and saved successfully")
+}

openapi/kb/addfile.go

@@ -6,26 +6,25 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/yaoapp/gou/graphrag/utils"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/exception"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/kun/maps"
 	"github.com/yaoapp/yao/attachment"
+	"github.com/yaoapp/yao/job"
 	"github.com/yaoapp/yao/kb"
 	"github.com/yaoapp/yao/openapi/response"
 )
 
-// AddFileProcess processes a file addition request with business logic only
+// CreateDocumentRecord creates a document record in the database immediately
-// This function is Gin-agnostic and can be used for both sync and async operations
+// This is called synchronously when the API request comes in
-func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) error {
+func CreateDocumentRecord(ctx context.Context, req *AddFileRequest, jobID string) error {
 	// Check if kb.Instance is available
 	if kb.Instance == nil {
 		return fmt.Errorf("knowledge base not initialized")
 	}
 
-	// Validate request
-	if err := req.Validate(); err != nil {
-		return err
-	}
-
 	// Get file manager
 	m, ok := attachment.Managers[req.Uploader]
 	if !ok {
@@ -49,11 +48,6 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e
 		return fmt.Errorf("failed to get file info: %w", err)
 	}
 
-	// DocID should be generated by the caller before calling this function
-	if req.DocID == "" {
-		return fmt.Errorf("document ID is required")
-	}
-
 	// Get KB config
 	config, err := kb.GetConfig()
 	if err != nil {
@@ -73,29 +67,52 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e
 		"file_path":      path,
 		"file_mime_type": contentType,
 		"size":           int64(fileInfo.Bytes),
-	}
+		"job_id":         jobID,
-
-	// Add job_id if provided (for async operations)
-	if len(jobID) > 0 && jobID[0] != "" {
-		documentData["job_id"] = jobID[0]
 	}
 
 	// Add base request fields
 	req.BaseUpsertRequest.AddBaseFields(documentData)
 
-	// First create database record
+	// Create database record
 	_, err = config.CreateDocument(maps.MapStrAny(documentData))
 	if err != nil {
 		return fmt.Errorf("failed to save document metadata: %w", err)
 	}
 
+	return nil
+}
+
+// HandleFileContent processes the actual file content and updates the knowledge base
+// This is called asynchronously by the job system
+func HandleFileContent(ctx context.Context, req *AddFileRequest) error {
+	// Check if kb.Instance is available
+	if kb.Instance == nil {
+		return fmt.Errorf("knowledge base not initialized")
+	}
+
+	// Get file manager
+	m, ok := attachment.Managers[req.Uploader]
+	if !ok {
+		return fmt.Errorf("invalid uploader: %s not found", req.Uploader)
+	}
+
+	// Get file info and path
+	path, contentType, err := m.LocalPath(ctx, req.FileID)
+	if err != nil {
+		return fmt.Errorf("failed to get local path: %w", err)
+	}
+
+	// Get KB config
+	config, err := kb.GetConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get KB config: %w", err)
+	}
+
 	// Convert request to UpsertOptions
 	upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions(path, contentType)
 	if err != nil {
-		// Rollback: remove the database record
+		// Update status to error
-		if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil {
+		config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()})
-			log.Error("Failed to rollback document database record: %v", rollbackErr)
-		}
 		return fmt.Errorf("failed to convert request to upsert options: %w", err)
 	}
 
@@ -134,10 +151,38 @@ func AddFileProcess(ctx context.Context, req *AddFileRequest, jobID ...string) e
 	return nil
 }
 
+// AddFileHandler processes a file addition request with business logic only
+// This function combines both document creation and content processing for sync operations
+func AddFileHandler(ctx context.Context, req *AddFileRequest, jobID ...string) error {
+	// Validate request
+	if err := req.Validate(); err != nil {
+		return err
+	}
+
+	// DocID should be generated by the caller before calling this function
+	if req.DocID == "" {
+		return fmt.Errorf("document ID is required")
+	}
+
+	// For sync operations, create document record and process content immediately
+	var jid string
+	if len(jobID) > 0 {
+		jid = jobID[0]
+	}
+
+	// Create document record
+	if err := CreateDocumentRecord(ctx, req, jid); err != nil {
+		return err
+	}
+
+	// Process file content
+	return HandleFileContent(ctx, req)
+}
+
 // addFileWithRequest processes a file addition with pre-parsed request using Gin context
 func addFileWithRequest(c *gin.Context, req *AddFileRequest) {
 	// Use the business logic function
-	err := AddFileProcess(c.Request.Context(), req)
+	err := AddFileHandler(c.Request.Context(), req)
 	if err != nil {
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrServerError.Code,
@@ -200,13 +245,17 @@ func AddFile(c *gin.Context) {
 func AddFileAsync(c *gin.Context) {
 	var req AddFileRequest
 
+	log.Info("AddFileAsync: Starting async file addition")
+
 	// Check if kb.Instance is available
 	if !checkKBInstance(c) {
+		log.Error("AddFileAsync: KB instance check failed")
 		return
 	}
 
 	// Parse and bind JSON request
 	if err := c.ShouldBindJSON(&req); err != nil {
+		log.Error("AddFileAsync: JSON binding failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: "Invalid request format: " + err.Error(),
@@ -215,8 +264,11 @@ func AddFileAsync(c *gin.Context) {
 		return
 	}
 
+	log.Info("AddFileAsync: Request parsed successfully: %+v", req)
+
 	// Validate request
 	if err := req.Validate(); err != nil {
+		log.Error("AddFileAsync: Request validation failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: err.Error(),
@@ -225,35 +277,324 @@ func AddFileAsync(c *gin.Context) {
 		return
 	}
 
+	log.Info("AddFileAsync: Request validation passed")
+
 	// Validate file and get path
 	_, _, err := validateFileAndGetPath(c, &req)
 	if err != nil {
+		log.Error("AddFileAsync: File validation failed: %v", err)
 		return
 	}
 
+	log.Info("AddFileAsync: File validation passed")
+
 	// Convert request to UpsertOptions (just for validation)
 	_, err = getUpsertOptions(c, &req.BaseUpsertRequest)
 	if err != nil {
+		log.Error("AddFileAsync: UpsertOptions validation failed: %v", err)
 		return
 	}
 
+	log.Info("AddFileAsync: UpsertOptions validation passed")
+
 	// Generate document ID if not provided
 	if req.DocID == "" {
 		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
 	}
 
-	// Create and run job
+	log.Info("AddFileAsync: Generated doc_id: %s", req.DocID)
-	job := NewJob()
+
-	jobID := job.Run(func() {
+	// Step 1: Create and save Job in one step to get JobID
-		err := AddFileProcess(context.Background(), &req, job.ID)
+	j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
-		if err != nil {
+		"name":        "KB Add File",
-			log.Error("Async file processing failed: %v", err)
+		"description": "Add file to knowledge base collection",
-		}
 	})
+	if err != nil {
+		log.Error("AddFileAsync: Job creation and save failed: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create and save job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddFileAsync: Job created and saved with ID: %s", j.JobID)
+
+	// Step 2: Create document record immediately with job_id
+	err = CreateDocumentRecord(c.Request.Context(), &req, j.JobID)
+	if err != nil {
+		log.Error("AddFileAsync: Failed to create document record: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create document record: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddFileAsync: Document record created successfully")
+
+	// Step 4: Add execution to job
+	jobData := map[string]interface{}{
+		"collection_id": req.CollectionID,
+		"file_id":       req.FileID,
+		"uploader":      req.Uploader,
+		"locale":        req.Locale,
+		"doc_id":        req.DocID,
+		"metadata":      req.Metadata,
+		"chunking":      req.Chunking,
+		"embedding":     req.Embedding,
+		"extraction":    req.Extraction,
+		"fetcher":       req.Fetcher,
+		"converter":     req.Converter,
+	}
+
+	err = j.Add(&job.ExecutionOptions{
+		Priority: 1,
+	}, "kb.documents.processfile", jobData)
+	if err != nil {
+		log.Error("AddFileAsync: Failed to add job execution: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to add job execution: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	// Step 5: Push the job to execution queue
+	err = j.Push()
+	if err != nil {
+		log.Error("AddFileAsync: Failed to push job: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to push job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddFileAsync: Job pushed successfully")
 
 	// Return job_id and doc_id
 	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
-		"job_id": jobID,
+		"job_id": j.JobID,
 		"doc_id": req.DocID,
 	})
 }
+
+// ProcessAddFile documents.addfile Knowledge Base add file processor
+// Args[0] map: Request parameters {"collection_id": "collection", "file_id": "file123", "uploader": "local", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessAddFile(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddFileRequest structure
+	req := parseAddFileRequest(reqMap)
+
+	// Get KB config to check if document exists
+	config, err := kb.GetConfig()
+	if err != nil {
+		exception.New("failed to get KB config: %s", 500, err.Error()).Throw()
+	}
+
+	// Check if document already exists
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{})
+	if err != nil || existingDoc == nil {
+		// Document doesn't exist, create it first (sync scenario)
+		log.Info("ProcessAddFile: Document %s not found, creating new record", req.DocID)
+
+		// Get job_id from request if provided (for async scenario)
+		var jobID string
+		if jid, ok := reqMap["job_id"].(string); ok {
+			jobID = jid
+		}
+
+		err = CreateDocumentRecord(ctx, req, jobID)
+		if err != nil {
+			exception.New("failed to create document record: %s", 500, err.Error()).Throw()
+		}
+	} else {
+		log.Info("ProcessAddFile: Document %s already exists, processing content only", req.DocID)
+	}
+
+	// Process file content
+	err = HandleFileContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process file: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// ProcessProcessFile documents.processfile Knowledge Base process file content processor (async version)
+// Args[0] map: Request parameters {"collection_id": "collection", "file_id": "file123", "uploader": "local", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessProcessFile(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddFileRequest structure
+	req := parseAddFileRequest(reqMap)
+
+	// This is async version - document record should already exist
+	// Just process the file content
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	err := HandleFileContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process file content: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// parseAddFileRequest parses request map into AddFileRequest structure
+func parseAddFileRequest(reqMap map[string]interface{}) *AddFileRequest {
+	req := &AddFileRequest{}
+
+	// Required fields
+	if collectionID, ok := reqMap["collection_id"].(string); ok {
+		req.CollectionID = collectionID
+	} else {
+		exception.New("collection_id is required", 400).Throw()
+	}
+
+	if fileID, ok := reqMap["file_id"].(string); ok {
+		req.FileID = fileID
+	} else {
+		exception.New("file_id is required", 400).Throw()
+	}
+
+	// Optional fields
+	if uploader, ok := reqMap["uploader"].(string); ok {
+		req.Uploader = uploader
+	} else {
+		req.Uploader = "local" // Default to local uploader
+	}
+
+	if locale, ok := reqMap["locale"].(string); ok {
+		req.Locale = locale
+	}
+
+	if docID, ok := reqMap["doc_id"].(string); ok {
+		req.DocID = docID
+	}
+
+	// Generate doc_id if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
+	// Handle metadata
+	if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok {
+		req.Metadata = metadata
+	}
+
+	// Handle chunking configuration
+	if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok {
+		chunking := &ProviderConfig{}
+		if providerID, ok := chunkingMap["provider_id"].(string); ok {
+			chunking.ProviderID = providerID
+		} else {
+			exception.New("chunking.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := chunkingMap["option_id"].(string); ok {
+			chunking.OptionID = optionID
+		}
+		req.Chunking = chunking
+	} else {
+		exception.New("chunking configuration is required", 400).Throw()
+	}
+
+	// Handle embedding configuration
+	if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok {
+		embedding := &ProviderConfig{}
+		if providerID, ok := embeddingMap["provider_id"].(string); ok {
+			embedding.ProviderID = providerID
+		} else {
+			exception.New("embedding.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := embeddingMap["option_id"].(string); ok {
+			embedding.OptionID = optionID
+		}
+		req.Embedding = embedding
+	} else {
+		exception.New("embedding configuration is required", 400).Throw()
+	}
+
+	// Handle optional extraction configuration
+	if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok {
+		extraction := &ProviderConfig{}
+		if providerID, ok := extractionMap["provider_id"].(string); ok {
+			extraction.ProviderID = providerID
+		}
+		if optionID, ok := extractionMap["option_id"].(string); ok {
+			extraction.OptionID = optionID
+		}
+		req.Extraction = extraction
+	}
+
+	// Handle optional fetcher configuration
+	if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok {
+		fetcher := &ProviderConfig{}
+		if providerID, ok := fetcherMap["provider_id"].(string); ok {
+			fetcher.ProviderID = providerID
+		}
+		if optionID, ok := fetcherMap["option_id"].(string); ok {
+			fetcher.OptionID = optionID
+		}
+		req.Fetcher = fetcher
+	}
+
+	// Handle optional converter configuration
+	if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok {
+		converter := &ProviderConfig{}
+		if providerID, ok := converterMap["provider_id"].(string); ok {
+			converter.ProviderID = providerID
+		}
+		if optionID, ok := converterMap["option_id"].(string); ok {
+			converter.OptionID = optionID
+		}
+		req.Converter = converter
+	}
+
+	return req
+}

openapi/kb/addtext.go

@@ -6,30 +6,24 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/yaoapp/gou/graphrag/utils"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/exception"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/kun/maps"
+	"github.com/yaoapp/yao/job"
 	"github.com/yaoapp/yao/kb"
 	"github.com/yaoapp/yao/openapi/response"
 )
 
-// AddTextProcess processes a text addition request with business logic only
+// CreateTextDocumentRecord creates a text document record in the database immediately
-// This function is Gin-agnostic and can be used for both sync and async operations
+// This is called synchronously when the API request comes in
-func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) error {
+func CreateTextDocumentRecord(ctx context.Context, req *AddTextRequest, jobID string) error {
 	// Check if kb.Instance is available
 	if kb.Instance == nil {
 		return fmt.Errorf("knowledge base not initialized")
 	}
 
-	// Validate request
-	if err := req.Validate(); err != nil {
-		return err
-	}
-
-	// DocID should be generated by the caller before calling this function
-	if req.DocID == "" {
-		return fmt.Errorf("document ID is required")
-	}
-
 	// Get KB config
 	config, err := kb.GetConfig()
 	if err != nil {
@@ -45,11 +39,7 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e
 		"status":        "pending",
 		"text_content":  req.Text,
 		"size":          int64(len(req.Text)),
-	}
+		"job_id":        jobID,
-
-	// Add job_id if provided (for async operations)
-	if len(jobID) > 0 && jobID[0] != "" {
-		documentData["job_id"] = jobID[0]
 	}
 
 	// Use title from metadata if available
@@ -62,19 +52,34 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e
 	// Add base request fields
 	req.BaseUpsertRequest.AddBaseFields(documentData)
 
-	// First create database record
+	// Create database record
 	_, err = config.CreateDocument(maps.MapStrAny(documentData))
 	if err != nil {
 		return fmt.Errorf("failed to save document metadata: %w", err)
 	}
 
+	return nil
+}
+
+// HandleTextContent processes the actual text content and updates the knowledge base
+// This is called asynchronously by the job system
+func HandleTextContent(ctx context.Context, req *AddTextRequest) error {
+	// Check if kb.Instance is available
+	if kb.Instance == nil {
+		return fmt.Errorf("knowledge base not initialized")
+	}
+
+	// Get KB config
+	config, err := kb.GetConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get KB config: %w", err)
+	}
+
 	// Convert request to UpsertOptions
 	upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions()
 	if err != nil {
-		// Rollback: remove the database record
+		// Update status to error
-		if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil {
+		config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()})
-			log.Error("Failed to rollback document database record: %v", rollbackErr)
-		}
 		return fmt.Errorf("failed to convert request to upsert options: %w", err)
 	}
 
@@ -113,10 +118,38 @@ func AddTextProcess(ctx context.Context, req *AddTextRequest, jobID ...string) e
 	return nil
 }
 
+// AddTextHandler processes a text addition request with business logic only
+// This function combines both document creation and content processing for sync operations
+func AddTextHandler(ctx context.Context, req *AddTextRequest, jobID ...string) error {
+	// Validate request
+	if err := req.Validate(); err != nil {
+		return err
+	}
+
+	// DocID should be generated by the caller before calling this function
+	if req.DocID == "" {
+		return fmt.Errorf("document ID is required")
+	}
+
+	// For sync operations, create document record and process content immediately
+	var jid string
+	if len(jobID) > 0 {
+		jid = jobID[0]
+	}
+
+	// Create document record
+	if err := CreateTextDocumentRecord(ctx, req, jid); err != nil {
+		return err
+	}
+
+	// Process text content
+	return HandleTextContent(ctx, req)
+}
+
 // addTextWithRequest processes a text addition with pre-parsed request using Gin context
 func addTextWithRequest(c *gin.Context, req *AddTextRequest) {
 	// Use the business logic function
-	err := AddTextProcess(c.Request.Context(), req)
+	err := AddTextHandler(c.Request.Context(), req)
 	if err != nil {
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrServerError.Code,
@@ -178,8 +211,17 @@ func AddText(c *gin.Context) {
 func AddTextAsync(c *gin.Context) {
 	var req AddTextRequest
 
+	log.Info("AddTextAsync: Starting async text addition")
+
+	// Check if kb.Instance is available
+	if !checkKBInstance(c) {
+		log.Error("AddTextAsync: KB instance check failed")
+		return
+	}
+
 	// Parse and bind JSON request
 	if err := c.ShouldBindJSON(&req); err != nil {
+		log.Error("AddTextAsync: JSON binding failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: "Invalid request format: " + err.Error(),
@@ -188,8 +230,11 @@ func AddTextAsync(c *gin.Context) {
 		return
 	}
 
+	log.Info("AddTextAsync: Request parsed successfully: %+v", req)
+
 	// Validate request
 	if err := req.Validate(); err != nil {
+		log.Error("AddTextAsync: Request validation failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: err.Error(),
@@ -198,34 +243,308 @@ func AddTextAsync(c *gin.Context) {
 		return
 	}
 
-	// Check if kb.Instance is available
+	log.Info("AddTextAsync: Request validation passed")
-	if !checkKBInstance(c) {
-		return
-	}
 
 	// Convert request to UpsertOptions (just for validation)
 	_, err := getUpsertOptions(c, &req.BaseUpsertRequest)
 	if err != nil {
+		log.Error("AddTextAsync: UpsertOptions validation failed: %v", err)
 		return
 	}
 
+	log.Info("AddTextAsync: UpsertOptions validation passed")
+
 	// Generate document ID if not provided
 	if req.DocID == "" {
 		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
 	}
 
-	// Create and run job
+	log.Info("AddTextAsync: Generated doc_id: %s", req.DocID)
-	job := NewJob()
+
-	jobID := job.Run(func() {
+	// Step 1: Create and save Job in one step to get JobID
-		err := AddTextProcess(context.Background(), &req, job.ID)
+	j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
-		if err != nil {
+		"name":        "KB Add Text",
-			log.Error("Async text processing failed: %v", err)
+		"description": "Add text to knowledge base collection",
-		}
 	})
+	if err != nil {
+		log.Error("AddTextAsync: Job creation and save failed: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create and save job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddTextAsync: Job created and saved with ID: %s", j.JobID)
+
+	// Step 2: Create document record immediately with job_id
+	err = CreateTextDocumentRecord(c.Request.Context(), &req, j.JobID)
+	if err != nil {
+		log.Error("AddTextAsync: Failed to create document record: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create document record: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddTextAsync: Document record created successfully")
+
+	// Step 3: Add execution to job
+	jobData := map[string]interface{}{
+		"collection_id": req.CollectionID,
+		"text":          req.Text,
+		"locale":        req.Locale,
+		"doc_id":        req.DocID,
+		"metadata":      req.Metadata,
+		"chunking":      req.Chunking,
+		"embedding":     req.Embedding,
+		"extraction":    req.Extraction,
+		"fetcher":       req.Fetcher,
+		"converter":     req.Converter,
+	}
+
+	err = j.Add(&job.ExecutionOptions{
+		Priority: 1,
+	}, "kb.documents.processtext", jobData)
+	if err != nil {
+		log.Error("AddTextAsync: Failed to add job execution: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to add job execution: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	// Step 4: Push the job to execution queue
+	err = j.Push()
+	if err != nil {
+		log.Error("AddTextAsync: Failed to push job: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to push job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddTextAsync: Job pushed successfully")
 
 	// Return job_id and doc_id
 	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
-		"job_id": jobID,
+		"job_id": j.JobID,
 		"doc_id": req.DocID,
 	})
 }
+
+// ProcessAddText documents.addtext Knowledge Base add text processor (sync version)
+// Args[0] map: Request parameters {"collection_id": "collection", "text": "content", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessAddText(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddTextRequest structure
+	req := parseAddTextRequest(reqMap)
+
+	// Get KB config to check if document exists
+	config, err := kb.GetConfig()
+	if err != nil {
+		exception.New("failed to get KB config: %s", 500, err.Error()).Throw()
+	}
+
+	// Check if document already exists
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{})
+	if err != nil || existingDoc == nil {
+		// Document doesn't exist, create it first (sync scenario)
+		log.Info("ProcessAddText: Document %s not found, creating new record", req.DocID)
+
+		// Get job_id from request if provided (for async scenario)
+		var jobID string
+		if jid, ok := reqMap["job_id"].(string); ok {
+			jobID = jid
+		}
+
+		err = CreateTextDocumentRecord(ctx, req, jobID)
+		if err != nil {
+			exception.New("failed to create document record: %s", 500, err.Error()).Throw()
+		}
+	} else {
+		log.Info("ProcessAddText: Document %s already exists, processing content only", req.DocID)
+	}
+
+	// Process text content
+	err = HandleTextContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process text: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// ProcessProcessText documents.processtext Knowledge Base process text content processor (async version)
+// Args[0] map: Request parameters {"collection_id": "collection", "text": "content", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessProcessText(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddTextRequest structure
+	req := parseAddTextRequest(reqMap)
+
+	// This is async version - document record should already exist
+	// Just process the text content
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	err := HandleTextContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process text content: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// parseAddTextRequest parses request map into AddTextRequest structure
+func parseAddTextRequest(reqMap map[string]interface{}) *AddTextRequest {
+	req := &AddTextRequest{}
+
+	// Required fields
+	if collectionID, ok := reqMap["collection_id"].(string); ok {
+		req.CollectionID = collectionID
+	} else {
+		exception.New("collection_id is required", 400).Throw()
+	}
+
+	if text, ok := reqMap["text"].(string); ok {
+		req.Text = text
+	} else {
+		exception.New("text is required", 400).Throw()
+	}
+
+	// Optional fields
+	if locale, ok := reqMap["locale"].(string); ok {
+		req.Locale = locale
+	}
+
+	if docID, ok := reqMap["doc_id"].(string); ok {
+		req.DocID = docID
+	}
+
+	// Generate doc_id if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
+	// Handle metadata
+	if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok {
+		req.Metadata = metadata
+	}
+
+	// Handle chunking configuration
+	if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok {
+		chunking := &ProviderConfig{}
+		if providerID, ok := chunkingMap["provider_id"].(string); ok {
+			chunking.ProviderID = providerID
+		} else {
+			exception.New("chunking.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := chunkingMap["option_id"].(string); ok {
+			chunking.OptionID = optionID
+		}
+		req.Chunking = chunking
+	} else {
+		exception.New("chunking configuration is required", 400).Throw()
+	}
+
+	// Handle embedding configuration
+	if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok {
+		embedding := &ProviderConfig{}
+		if providerID, ok := embeddingMap["provider_id"].(string); ok {
+			embedding.ProviderID = providerID
+		} else {
+			exception.New("embedding.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := embeddingMap["option_id"].(string); ok {
+			embedding.OptionID = optionID
+		}
+		req.Embedding = embedding
+	} else {
+		exception.New("embedding configuration is required", 400).Throw()
+	}
+
+	// Handle optional extraction configuration
+	if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok {
+		extraction := &ProviderConfig{}
+		if providerID, ok := extractionMap["provider_id"].(string); ok {
+			extraction.ProviderID = providerID
+		}
+		if optionID, ok := extractionMap["option_id"].(string); ok {
+			extraction.OptionID = optionID
+		}
+		req.Extraction = extraction
+	}
+
+	// Handle optional fetcher configuration
+	if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok {
+		fetcher := &ProviderConfig{}
+		if providerID, ok := fetcherMap["provider_id"].(string); ok {
+			fetcher.ProviderID = providerID
+		}
+		if optionID, ok := fetcherMap["option_id"].(string); ok {
+			fetcher.OptionID = optionID
+		}
+		req.Fetcher = fetcher
+	}
+
+	// Handle optional converter configuration
+	if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok {
+		converter := &ProviderConfig{}
+		if providerID, ok := converterMap["provider_id"].(string); ok {
+			converter.ProviderID = providerID
+		}
+		if optionID, ok := converterMap["option_id"].(string); ok {
+			converter.OptionID = optionID
+		}
+		req.Converter = converter
+	}
+
+	return req
+}

openapi/kb/addurl.go

@@ -6,30 +6,24 @@ import (
 
 	"github.com/gin-gonic/gin"
 	"github.com/yaoapp/gou/graphrag/utils"
+	"github.com/yaoapp/gou/model"
+	"github.com/yaoapp/gou/process"
+	"github.com/yaoapp/kun/exception"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/kun/maps"
+	"github.com/yaoapp/yao/job"
 	"github.com/yaoapp/yao/kb"
 	"github.com/yaoapp/yao/openapi/response"
 )
 
-// AddURLProcess processes a URL addition request with business logic only
+// CreateURLDocumentRecord creates a URL document record in the database immediately
-// This function is Gin-agnostic and can be used for both sync and async operations
+// This is called synchronously when the API request comes in
-func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) error {
+func CreateURLDocumentRecord(ctx context.Context, req *AddURLRequest, jobID string) error {
 	// Check if kb.Instance is available
 	if kb.Instance == nil {
 		return fmt.Errorf("knowledge base not initialized")
 	}
 
-	// Validate request
-	if err := req.Validate(); err != nil {
-		return err
-	}
-
-	// DocID should be generated by the caller before calling this function
-	if req.DocID == "" {
-		return fmt.Errorf("document ID is required")
-	}
-
 	// Get KB config
 	config, err := kb.GetConfig()
 	if err != nil {
@@ -44,11 +38,7 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err
 		"type":          "url",
 		"status":        "pending",
 		"url":           req.URL,
-	}
+		"job_id":        jobID,
-
-	// Add job_id if provided (for async operations)
-	if len(jobID) > 0 && jobID[0] != "" {
-		documentData["job_id"] = jobID[0]
 	}
 
 	// Use title from metadata if available
@@ -61,19 +51,34 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err
 	// Add base request fields
 	req.BaseUpsertRequest.AddBaseFields(documentData)
 
-	// First create database record
+	// Create database record
 	_, err = config.CreateDocument(maps.MapStrAny(documentData))
 	if err != nil {
 		return fmt.Errorf("failed to save document metadata: %w", err)
 	}
 
+	return nil
+}
+
+// HandleURLContent processes the actual URL content and updates the knowledge base
+// This is called asynchronously by the job system
+func HandleURLContent(ctx context.Context, req *AddURLRequest) error {
+	// Check if kb.Instance is available
+	if kb.Instance == nil {
+		return fmt.Errorf("knowledge base not initialized")
+	}
+
+	// Get KB config
+	config, err := kb.GetConfig()
+	if err != nil {
+		return fmt.Errorf("failed to get KB config: %w", err)
+	}
+
 	// Convert request to UpsertOptions
 	upsertOptions, err := req.BaseUpsertRequest.ToUpsertOptions()
 	if err != nil {
-		// Rollback: remove the database record
+		// Update status to error
-		if rollbackErr := config.RemoveDocument(req.DocID); rollbackErr != nil {
+		config.UpdateDocument(req.DocID, maps.MapStrAny{"status": "error", "error_message": err.Error()})
-			log.Error("Failed to rollback document database record: %v", rollbackErr)
-		}
 		return fmt.Errorf("failed to convert request to upsert options: %w", err)
 	}
 
@@ -112,10 +117,38 @@ func AddURLProcess(ctx context.Context, req *AddURLRequest, jobID ...string) err
 	return nil
 }
 
+// AddURLHandler processes a URL addition request with business logic only
+// This function combines both document creation and content processing for sync operations
+func AddURLHandler(ctx context.Context, req *AddURLRequest, jobID ...string) error {
+	// Validate request
+	if err := req.Validate(); err != nil {
+		return err
+	}
+
+	// DocID should be generated by the caller before calling this function
+	if req.DocID == "" {
+		return fmt.Errorf("document ID is required")
+	}
+
+	// For sync operations, create document record and process content immediately
+	var jid string
+	if len(jobID) > 0 {
+		jid = jobID[0]
+	}
+
+	// Create document record
+	if err := CreateURLDocumentRecord(ctx, req, jid); err != nil {
+		return err
+	}
+
+	// Process URL content
+	return HandleURLContent(ctx, req)
+}
+
 // addURLWithRequest processes a URL addition with pre-parsed request using Gin context
 func addURLWithRequest(c *gin.Context, req *AddURLRequest) {
 	// Use the business logic function
-	err := AddURLProcess(c.Request.Context(), req)
+	err := AddURLHandler(c.Request.Context(), req)
 	if err != nil {
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrServerError.Code,
@@ -178,8 +211,17 @@ func AddURL(c *gin.Context) {
 func AddURLAsync(c *gin.Context) {
 	var req AddURLRequest
 
+	log.Info("AddURLAsync: Starting async URL addition")
+
+	// Check if kb.Instance is available
+	if !checkKBInstance(c) {
+		log.Error("AddURLAsync: KB instance check failed")
+		return
+	}
+
 	// Parse and bind JSON request
 	if err := c.ShouldBindJSON(&req); err != nil {
+		log.Error("AddURLAsync: JSON binding failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: "Invalid request format: " + err.Error(),
@@ -188,8 +230,11 @@ func AddURLAsync(c *gin.Context) {
 		return
 	}
 
+	log.Info("AddURLAsync: Request parsed successfully: %+v", req)
+
 	// Validate request
 	if err := req.Validate(); err != nil {
+		log.Error("AddURLAsync: Request validation failed: %v", err)
 		errorResp := &response.ErrorResponse{
 			Code:             response.ErrInvalidRequest.Code,
 			ErrorDescription: err.Error(),
@@ -198,34 +243,308 @@ func AddURLAsync(c *gin.Context) {
 		return
 	}
 
-	// Check if kb.Instance is available
+	log.Info("AddURLAsync: Request validation passed")
-	if !checkKBInstance(c) {
-		return
-	}
 
 	// Convert request to UpsertOptions (just for validation)
 	_, err := getUpsertOptions(c, &req.BaseUpsertRequest)
 	if err != nil {
+		log.Error("AddURLAsync: UpsertOptions validation failed: %v", err)
 		return
 	}
 
+	log.Info("AddURLAsync: UpsertOptions validation passed")
+
 	// Generate document ID if not provided
 	if req.DocID == "" {
 		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
 	}
 
-	// Create and run job
+	log.Info("AddURLAsync: Generated doc_id: %s", req.DocID)
-	job := NewJob()
+
-	jobID := job.Run(func() {
+	// Step 1: Create and save Job in one step to get JobID
-		err := AddURLProcess(context.Background(), &req, job.ID)
+	j, err := job.OnceAndSave(job.GOROUTINE, map[string]interface{}{
-		if err != nil {
+		"name":        "KB Add URL",
-			log.Error("Async URL processing failed: %v", err)
+		"description": "Add URL to knowledge base collection",
-		}
 	})
+	if err != nil {
+		log.Error("AddURLAsync: Job creation and save failed: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create and save job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddURLAsync: Job created and saved with ID: %s", j.JobID)
+
+	// Step 2: Create document record immediately with job_id
+	err = CreateURLDocumentRecord(c.Request.Context(), &req, j.JobID)
+	if err != nil {
+		log.Error("AddURLAsync: Failed to create document record: %v", err)
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to create document record: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddURLAsync: Document record created successfully")
+
+	// Step 3: Add execution to job
+	jobData := map[string]interface{}{
+		"collection_id": req.CollectionID,
+		"url":           req.URL,
+		"locale":        req.Locale,
+		"doc_id":        req.DocID,
+		"metadata":      req.Metadata,
+		"chunking":      req.Chunking,
+		"embedding":     req.Embedding,
+		"extraction":    req.Extraction,
+		"fetcher":       req.Fetcher,
+		"converter":     req.Converter,
+	}
+
+	err = j.Add(&job.ExecutionOptions{
+		Priority: 1,
+	}, "kb.documents.processurl", jobData)
+	if err != nil {
+		log.Error("AddURLAsync: Failed to add job execution: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to add job execution: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	// Step 4: Push the job to execution queue
+	err = j.Push()
+	if err != nil {
+		log.Error("AddURLAsync: Failed to push job: %v", err)
+		// Rollback: remove document record
+		if config, err := kb.GetConfig(); err == nil {
+			config.RemoveDocument(req.DocID)
+		}
+		errorResp := &response.ErrorResponse{
+			Code:             response.ErrServerError.Code,
+			ErrorDescription: "Failed to push job: " + err.Error(),
+		}
+		response.RespondWithError(c, response.StatusInternalServerError, errorResp)
+		return
+	}
+
+	log.Info("AddURLAsync: Job pushed successfully")
 
 	// Return job_id and doc_id
 	response.RespondWithSuccess(c, response.StatusCreated, gin.H{
-		"job_id": jobID,
+		"job_id": j.JobID,
 		"doc_id": req.DocID,
 	})
 }
+
+// ProcessAddURL documents.addurl Knowledge Base add URL processor (sync version)
+// Args[0] map: Request parameters {"collection_id": "collection", "url": "https://example.com", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessAddURL(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddURLRequest structure
+	req := parseAddURLRequest(reqMap)
+
+	// Get KB config to check if document exists
+	config, err := kb.GetConfig()
+	if err != nil {
+		exception.New("failed to get KB config: %s", 500, err.Error()).Throw()
+	}
+
+	// Check if document already exists
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+
+	existingDoc, err := config.FindDocument(req.DocID, model.QueryParam{})
+	if err != nil || existingDoc == nil {
+		// Document doesn't exist, create it first (sync scenario)
+		log.Info("ProcessAddURL: Document %s not found, creating new record", req.DocID)
+
+		// Get job_id from request if provided (for async scenario)
+		var jobID string
+		if jid, ok := reqMap["job_id"].(string); ok {
+			jobID = jid
+		}
+
+		err = CreateURLDocumentRecord(ctx, req, jobID)
+		if err != nil {
+			exception.New("failed to create document record: %s", 500, err.Error()).Throw()
+		}
+	} else {
+		log.Info("ProcessAddURL: Document %s already exists, processing content only", req.DocID)
+	}
+
+	// Process URL content
+	err = HandleURLContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process URL: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// ProcessProcessURL documents.processurl Knowledge Base process URL content processor (async version)
+// Args[0] map: Request parameters {"collection_id": "collection", "url": "https://example.com", ...}
+// Return: map: Response data {"doc_id": "document_id"}
+func ProcessProcessURL(process *process.Process) interface{} {
+	process.ValidateArgNums(1)
+
+	// Get parameters
+	reqMap := process.ArgsMap(0)
+
+	// Check knowledge base instance
+	if kb.Instance == nil {
+		exception.New("knowledge base not initialized", 500).Throw()
+	}
+
+	// Convert parameters to AddURLRequest structure
+	req := parseAddURLRequest(reqMap)
+
+	// This is async version - document record should already exist
+	// Just process the URL content
+	ctx := process.Context
+	if ctx == nil {
+		ctx = context.Background()
+	}
+	err := HandleURLContent(ctx, req)
+	if err != nil {
+		exception.New("failed to process URL content: %s", 500, err.Error()).Throw()
+	}
+
+	// Return result
+	return maps.MapStrAny{
+		"doc_id": req.DocID,
+	}
+}
+
+// parseAddURLRequest parses request map into AddURLRequest structure
+func parseAddURLRequest(reqMap map[string]interface{}) *AddURLRequest {
+	req := &AddURLRequest{}
+
+	// Required fields
+	if collectionID, ok := reqMap["collection_id"].(string); ok {
+		req.CollectionID = collectionID
+	} else {
+		exception.New("collection_id is required", 400).Throw()
+	}
+
+	if url, ok := reqMap["url"].(string); ok {
+		req.URL = url
+	} else {
+		exception.New("url is required", 400).Throw()
+	}
+
+	// Optional fields
+	if locale, ok := reqMap["locale"].(string); ok {
+		req.Locale = locale
+	}
+
+	if docID, ok := reqMap["doc_id"].(string); ok {
+		req.DocID = docID
+	}
+
+	// Generate doc_id if not provided
+	if req.DocID == "" {
+		req.DocID = utils.GenDocIDWithCollectionID(req.CollectionID)
+	}
+
+	// Handle metadata
+	if metadata, ok := reqMap["metadata"].(map[string]interface{}); ok {
+		req.Metadata = metadata
+	}
+
+	// Handle chunking configuration
+	if chunkingMap, ok := reqMap["chunking"].(map[string]interface{}); ok {
+		chunking := &ProviderConfig{}
+		if providerID, ok := chunkingMap["provider_id"].(string); ok {
+			chunking.ProviderID = providerID
+		} else {
+			exception.New("chunking.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := chunkingMap["option_id"].(string); ok {
+			chunking.OptionID = optionID
+		}
+		req.Chunking = chunking
+	} else {
+		exception.New("chunking configuration is required", 400).Throw()
+	}
+
+	// Handle embedding configuration
+	if embeddingMap, ok := reqMap["embedding"].(map[string]interface{}); ok {
+		embedding := &ProviderConfig{}
+		if providerID, ok := embeddingMap["provider_id"].(string); ok {
+			embedding.ProviderID = providerID
+		} else {
+			exception.New("embedding.provider_id is required", 400).Throw()
+		}
+		if optionID, ok := embeddingMap["option_id"].(string); ok {
+			embedding.OptionID = optionID
+		}
+		req.Embedding = embedding
+	} else {
+		exception.New("embedding configuration is required", 400).Throw()
+	}
+
+	// Handle optional extraction configuration
+	if extractionMap, ok := reqMap["extraction"].(map[string]interface{}); ok {
+		extraction := &ProviderConfig{}
+		if providerID, ok := extractionMap["provider_id"].(string); ok {
+			extraction.ProviderID = providerID
+		}
+		if optionID, ok := extractionMap["option_id"].(string); ok {
+			extraction.OptionID = optionID
+		}
+		req.Extraction = extraction
+	}
+
+	// Handle optional fetcher configuration
+	if fetcherMap, ok := reqMap["fetcher"].(map[string]interface{}); ok {
+		fetcher := &ProviderConfig{}
+		if providerID, ok := fetcherMap["provider_id"].(string); ok {
+			fetcher.ProviderID = providerID
+		}
+		if optionID, ok := fetcherMap["option_id"].(string); ok {
+			fetcher.OptionID = optionID
+		}
+		req.Fetcher = fetcher
+	}
+
+	// Handle optional converter configuration
+	if converterMap, ok := reqMap["converter"].(map[string]interface{}); ok {
+		converter := &ProviderConfig{}
+		if providerID, ok := converterMap["provider_id"].(string); ok {
+			converter.ProviderID = providerID
+		}
+		if optionID, ok := converterMap["option_id"].(string); ok {
+			converter.OptionID = optionID
+		}
+		req.Converter = converter
+	}
+
+	return req
+}

openapi/kb/document.go

@@ -6,7 +6,6 @@ import (
 	"strings"
 
 	"github.com/gin-gonic/gin"
-	"github.com/google/uuid"
 	"github.com/yaoapp/gou/graphrag/types"
 	"github.com/yaoapp/gou/model"
 	"github.com/yaoapp/yao/attachment"
@@ -52,26 +51,6 @@ var (
 	}
 )
 
-// SimpleJob represents a simple job for async operations
-// TODO: replace with proper job system later
-type SimpleJob struct {
-	ID string
-}
-
-// NewJob creates a new simple job
-func NewJob() *SimpleJob {
-	return &SimpleJob{
-		ID: uuid.New().String(),
-	}
-}
-
-// Run executes the job function asynchronously and returns job ID
-func (j *SimpleJob) Run(fn func()) string {
-	// temporary solution to handle async operations ( TODO: use job queue )
-	go fn()
-	return j.ID
-}
-
 // Document Management Handlers
 
 // ListDocuments lists documents with pagination

openapi/kb/graph.go

@@ -471,21 +471,13 @@ func ExtractSegmentGraphAsync(c *gin.Context) {
 
 	// TODO: Implement document permission validation for docID
 
-	// Create and run job
+	// TODO: Implement async extract segment graph logic using Job system
-	job := NewJob()
-	jobID := job.Run(func() {
-		// TODO: Implement async extract segment graph logic
 	// err := ExtractSegmentGraphProcess(context.Background(), segmentID, extractOptions, job.ID)
-		// For now, just simulate async processing
-		// if err != nil {
-		//     log.Error("Async graph extraction failed: %v", err)
-		// }
-	})
 
-	// Return job ID for status tracking
+	// Temporary response until async implementation is completed
 	result := gin.H{
-		"job_id":     jobID,
+		"message":    "Async graph extraction not yet implemented",
-		"message":    "Graph extraction started",
+		"status":     "pending_implementation",
 		"doc_id":     docID,
 		"segment_id": segmentID,
 	}

openapi/kb/kb.go

@@ -2,11 +2,24 @@ package kb
 
 import (
 	"github.com/gin-gonic/gin"
+	"github.com/yaoapp/gou/process"
 	"github.com/yaoapp/kun/log"
 	"github.com/yaoapp/yao/kb"
 	"github.com/yaoapp/yao/openapi/oauth/types"
 )
 
+func init() {
+	// Register kb process handlers
+	process.RegisterGroup("kb", map[string]process.Handler{
+		"documents.addfile":     ProcessAddFile,
+		"documents.addtext":     ProcessAddText,
+		"documents.addurl":      ProcessAddURL,
+		"documents.processfile": ProcessProcessFile,
+		"documents.processtext": ProcessProcessText,
+		"documents.processurl":  ProcessProcessURL,
+	})
+}
+
 // Attach attaches the Knowledge Base API to the router
 func Attach(group *gin.RouterGroup, oauth types.OAuth) {
 

openapi/kb/segment.go

@@ -671,22 +671,13 @@ func AddSegmentsAsync(c *gin.Context) {
 
 	// TODO: Implement document permission validation for docID
 
-	// Create and run job
+	// TODO: Implement async add segments logic using Job system
-	job := NewJob()
-	jobID := job.Run(func() {
-		// TODO: Implement async add segments logic
 	// This should call the same logic as AddSegments but in background
-		// err := AddSegmentsProcess(context.Background(), &req, job.ID)
-		// For now, just simulate async processing
-		// if err != nil {
-		//     log.Error("Async segments addition failed: %v", err)
-		// }
-	})
 
-	// Return job ID for status tracking
+	// Temporary response until async implementation is completed
 	result := gin.H{
-		"job_id":  jobID,
+		"message": "Async segments addition not yet implemented",
-		"message": "Segments addition started",
+		"status":  "pending_implementation",
 		"doc_id":  docID,
 	}
 
@@ -730,22 +721,13 @@ func UpdateSegmentsAsync(c *gin.Context) {
 	// TODO: Validate request body
 	// TODO: Implement document permission validation for docID
 
-	// Create and run job
+	// TODO: Implement async update segments logic using Job system
-	job := NewJob()
-	jobID := job.Run(func() {
-		// TODO: Implement async update segments logic
 	// This should call the same logic as UpdateSegments but in background
-		// err := UpdateSegmentsProcess(context.Background(), docID, requestBody, job.ID)
-		// For now, just simulate async processing
-		// if err != nil {
-		//     log.Error("Async segments update failed: %v", err)
-		// }
-	})
 
-	// Return job ID for status tracking
+	// Temporary response until async implementation is completed
 	result := gin.H{
-		"job_id":  jobID,
+		"message": "Async segments update not yet implemented",
-		"message": "Segments update started",
+		"status":  "pending_implementation",
 		"doc_id":  docID,
 	}