Go: align cancel-parse error with Python for un-started documents (#18245)

internal/service/document/document_ingest.go

@@ -105,6 +105,10 @@ func (s *DocumentService) Ingest(ctx context.Context, userID string, req *Ingest
 		if run == string(entity.TaskStatusCancel) {
 			if err = s.CancelDocParse(ctx, doc); err != nil {
 				common.Error(fmt.Sprintf("go side, start to process %s, run is cancel", doc.ID), err)
+				if errors.Is(err, errParseNotRunning) {
+					// Mirror the Python /documents/ingest endpoint's message.
+					return common.CodeDataError, errors.New("Cannot cancel a task that is not in RUNNING status")
+				}
 				return common.CodeDataError, err
 			}
 			if err = s.documentDAO.UpdateByID(ctx, dao.DB, doc.ID, map[string]interface{}{

internal/service/document/document_parse.go

@@ -273,7 +273,12 @@ func (s *DocumentService) StopParseDocuments(ctx context.Context, datasetID stri
 	successCount := 0
 	for _, doc := range docs {
 		if cancelErr := s.CancelDocParse(ctx, doc); cancelErr != nil {
-			errs = append(errs, cancelErr.Error())
+			if errors.Is(cancelErr, errParseNotRunning) {
+				// Mirror the Python /documents/stop endpoint's message.
+				errs = append(errs, "Can't stop parsing document that has not started or already completed")
+			} else {
+				errs = append(errs, cancelErr.Error())
+			}
 			continue
 		}
 		successCount++
@@ -330,19 +335,38 @@ func (s *DocumentService) validateDocsInDataset(ctx context.Context, docIDs []st
 	return docs, nil
 }
 
+// errParseNotRunning is returned by CancelDocParse when the document is not in
+// a cancelable state: its run status is neither RUNNING nor CANCEL and it has
+// no in-flight ingestion task. Callers map it to their endpoint-specific
+// message (the Python /documents/ingest and /documents/stop messages differ).
+var errParseNotRunning = errors.New("parse task is not in running status")
+
 // CancelDocParse stops the ingestion task for the document by calling
 // RequestStop (STOPPING), then marks the document run status as CANCEL.
+// It mirrors the Python cancel precondition: only a document whose run status
+// is RUNNING or CANCEL, or one with an in-flight ingestion task
+// (CREATED/RUNNING/STOPPING), can be canceled; otherwise errParseNotRunning
+// is returned. A missing ingestion task is not an error by itself — cancel is
+// then a no-op on the task side, matching Python's cancel_all_task_of.
 func (s *DocumentService) CancelDocParse(ctx context.Context, doc *entity.Document) error {
 	task, err := s.ingestionTaskDAO.GetByDocumentID(ctx, dao.DB, doc.ID)
 	if err != nil {
 		return fmt.Errorf("failed to get ingestion task for %s: %v", doc.ID, err)
 	}
-	if task == nil {
-		return fmt.Errorf("no ingestion task found for document %s", doc.ID)
+
+	docRun := ""
+	if doc.Run != nil {
+		docRun = *doc.Run
+	}
+	inFlight := task != nil && (task.Status == common.CREATED || task.Status == common.RUNNING || task.Status == common.STOPPING)
+	if docRun != string(entity.TaskStatusRunning) && docRun != string(entity.TaskStatusCancel) && !inFlight {
+		return errParseNotRunning
 	}
 
-	if _, err = s.ingestionTaskSvc.RequestStop(ctx, task.ID); err != nil {
-		return fmt.Errorf("failed to stop ingestion task %s: %v", task.ID, err)
+	if task != nil {
+		if _, err = s.ingestionTaskSvc.RequestStop(ctx, task.ID); err != nil {
+			return fmt.Errorf("failed to stop ingestion task %s: %v", task.ID, err)
+		}
 	}
 
 	if upErr := s.documentDAO.UpdateByID(ctx, dao.DB, doc.ID, map[string]interface{}{"run": string(entity.TaskStatusCancel)}); upErr != nil {

internal/service/document/document_test.go

@@ -1331,6 +1331,9 @@ func TestStopParseDocuments_NotRunningOrCancel(t *testing.T) {
 	if !ok || len(errors) == 0 {
 		t.Fatal("expected errors in result")
 	}
+	if errors[0] != "Can't stop parsing document that has not started or already completed" {
+		t.Fatalf("unexpected error message: %q", errors[0])
+	}
 }
 
 func TestStopParseDocuments_UnfinishedTask(t *testing.T) {
@@ -2978,6 +2981,116 @@ func TestIngest_CancelDoesNotDeleteIngestionTask(t *testing.T) {
 	}
 }
 
+// TestIngest_CancelUnstartedDocument mirrors the Python /documents/ingest
+// behavior: canceling a document whose parse never started (run=UNSTART, no
+// ingestion task) must be rejected with code 102 and the Python message, not
+// an internal "no ingestion task found" error.
+func TestIngest_CancelUnstartedDocument(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertUserTenantForAccessCheck(t, "user-1", "tenant-1")
+	insertTestKB(t, "kb-1", "tenant-1", 0, 0, 0)
+	insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusUnstart), 10, 5)
+
+	svc := testDocumentService(t)
+	ctx := t.Context()
+	code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{
+		DocIDs: []string{"doc-1"},
+		Run:    string(entity.TaskStatusCancel),
+	})
+	if err == nil {
+		t.Fatal("expected error when canceling an un-started document")
+	}
+	if code != common.CodeDataError {
+		t.Fatalf("expected code %v, got %v", common.CodeDataError, code)
+	}
+	if err.Error() != "Cannot cancel a task that is not in RUNNING status" {
+		t.Fatalf("unexpected error message: %q", err.Error())
+	}
+}
+
+// TestIngest_CancelCompletedDocument: canceling an already finished parse
+// (run=DONE, ingestion task COMPLETED) is rejected with the same Python
+// message as the un-started case.
+func TestIngest_CancelCompletedDocument(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertUserTenantForAccessCheck(t, "user-1", "tenant-1")
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusDone), 10, 5)
+	insertTestIngestionTaskWithStatus(t, "task-1", "user-1", "doc-1", "kb-1", common.COMPLETED)
+
+	svc := testDocumentService(t)
+	ctx := t.Context()
+	code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{
+		DocIDs: []string{"doc-1"},
+		Run:    string(entity.TaskStatusCancel),
+	})
+	if err == nil {
+		t.Fatal("expected error when canceling a completed document")
+	}
+	if code != common.CodeDataError {
+		t.Fatalf("expected code %v, got %v", common.CodeDataError, code)
+	}
+	if err.Error() != "Cannot cancel a task that is not in RUNNING status" {
+		t.Fatalf("unexpected error message: %q", err.Error())
+	}
+}
+
+// TestIngest_CancelUnstartedWithInFlightTask: run=UNSTART but an ingestion
+// task was just enqueued (CREATED) — cancel is allowed, matching the Python
+// has_unfinished_task branch.
+func TestIngest_CancelUnstartedWithInFlightTask(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertUserTenantForAccessCheck(t, "user-1", "tenant-1")
+	insertTestKB(t, "kb-1", "tenant-1", 0, 0, 0)
+	insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusUnstart), 10, 5)
+	insertTestIngestionTask(t, "task-1", "user-1", "doc-1", "kb-1")
+
+	svc := testDocumentService(t)
+	ctx := t.Context()
+	code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{
+		DocIDs: []string{"doc-1"},
+		Run:    string(entity.TaskStatusCancel),
+	})
+	if err != nil {
+		t.Fatalf("Ingest(cancel) with in-flight task: %v", err)
+	}
+	if code != common.CodeSuccess {
+		t.Fatalf("expected code %v, got %v", common.CodeSuccess, code)
+	}
+
+	doc, _ := dao.NewDocumentDAO().GetByID(ctx, db, "doc-1")
+	if doc == nil || doc.Run == nil || *doc.Run != string(entity.TaskStatusCancel) {
+		t.Fatalf("expected doc run=CANCEL, got %v", doc.Run)
+	}
+}
+
+// TestIngest_CancelAgainWithoutTask: re-canceling a document already in
+// CANCEL state is accepted even when its ingestion task is gone — Python
+// treats run=CANCEL as cancelable and cancel_all_task_of is a no-op.
+func TestIngest_CancelAgainWithoutTask(t *testing.T) {
+	db := setupServiceTestDB(t)
+	pushServiceDB(t, db)
+	insertUserTenantForAccessCheck(t, "user-1", "tenant-1")
+	insertTestKB(t, "kb-1", "tenant-1", 1, 10, 5)
+	insertTestDocWithRun(t, "doc-1", "kb-1", string(entity.TaskStatusCancel), 10, 5)
+
+	svc := testDocumentService(t)
+	ctx := t.Context()
+	code, err := svc.Ingest(ctx, "user-1", &IngestDocumentRequest{
+		DocIDs: []string{"doc-1"},
+		Run:    string(entity.TaskStatusCancel),
+	})
+	if err != nil {
+		t.Fatalf("Ingest(re-cancel) without task: %v", err)
+	}
+	if code != common.CodeSuccess {
+		t.Fatalf("expected code %v, got %v", common.CodeSuccess, code)
+	}
+}
+
 func TestUpdateRunProgressMirrorsFields(t *testing.T) {
 	db := setupServiceTestDB(t)
 	pushServiceDB(t, db)