fix(import): resume failed imports and skip finished or half-parsed ones

frontend/src/pages/ImportDetail.vue

@@ -127,18 +127,18 @@ function confirmReclassify() {
 	});
 }
 
-// A killed worker leaves the import in Parsing/Remediating with no job behind it.
+// A killed or timed-out job leaves the import stopped with no job behind it.
 const canResume = useCall({
 	url: "/api/v2/method/wikify.api.imports.can_resume_import",
 	method: "GET",
 	immediate: false,
 });
-const isRunningStatus = computed(() => ["Parsing", "Remediating"].includes(status.value));
-const isStuck = computed(() => isRunningStatus.value && canResume.data === true);
+const mayBeResumable = computed(() => ["Parsing", "Remediating", "Failed"].includes(status.value));
+const isStuck = computed(() => mayBeResumable.value && canResume.data === true);
 watch(
 	status,
 	() => {
-		if (isRunningStatus.value) canResume.submit({ import_name: props.name });
+		if (mayBeResumable.value) canResume.submit({ import_name: props.name });
 	},
 	{ immediate: true }
 );
@@ -149,8 +149,12 @@ const resume = useCall({
 });
 async function resumeImport() {
 	await resume.submit({ import_name: props.name });
-	if (resume.error) toast.error(resume.error?.messages?.[0] || "Could not resume the import");
-	await canResume.submit({ import_name: props.name });
+	if (resume.error) {
+		toast.error(resume.error?.messages?.[0] || "Could not resume the import");
+		await canResume.submit({ import_name: props.name });
+	} else {
+		imp.reload();
+	}
 }
 
 // Attach the document (with its project) as the agent's default context. Keyed on the

wikify/api/imports.py

@@ -6,12 +6,13 @@
 
 from wikify.api.sections import assert_not_published
 from wikify.engine import preview_wiki as _preview_wiki
+from wikify.engine import store
 from wikify.jobs import generate as generate_job
 from wikify.jobs._util import IMPORT_JOB_TIMEOUT, import_job_id, log, publish_progress
 from wikify.seed import seed_uncategorized_project
 
 MAX_BATCH = 25
-RESUMABLE_STATUSES = ("Parsing", "Remediating")
+RESUMABLE_STATUSES = ("Parsing", "Remediating", "Failed")
 
 
 def assert_readable_file(file_url: str) -> None:
@@ -44,7 +45,9 @@ def enqueue_parse(import_name: str) -> None:
 
 
 def is_stuck(import_doc) -> bool:
-	return import_doc.status in RESUMABLE_STATUSES and not is_job_enqueued(import_job_id(import_doc.name))
+	if import_doc.status not in RESUMABLE_STATUSES or is_job_enqueued(import_job_id(import_doc.name)):
+		return False
+	return store.needs_resume(import_doc.source_document)
 
 
 @frappe.whitelist(methods=["POST"])
@@ -114,6 +117,7 @@ def resume_import(import_name: str) -> str:
 			)
 		)
 
+	import_doc.db_set({"status": "Queued", "error": None})
 	enqueue_parse(import_name)
 	return import_name
 

wikify/engine/store.py

@@ -151,6 +151,14 @@ def set_canonical(page_name: str, markdown: str, composite: float | None, source
 	invalidate_page(page_name)
 
 
+def needs_resume(source_document: str | None) -> bool:
+	return not source_document or bool(
+		frappe.db.exists(
+			"Source Page", {"source_document": source_document, "canonical_source": ("is", "not set")}
+		)
+	)
+
+
 def set_canonical_mean(source_document: str, mean: float | None) -> None:
 	frappe.db.set_value("Source Document", source_document, "canonical_mean", mean)
 

wikify/jobs/parse.py

@@ -52,6 +52,8 @@ def page_cb(page_no, total, kind, score, metrics) -> None:
 		# Parse + score only; the tree is built after remediation (below) so it's built once,
 		# over the cleaned canonical markdown rather than the raw baseline.
 		resuming = bool(imp.source_document)
+		if not resuming:
+			delete_partial_documents(import_name)
 		source_document = imp.source_document or parse_pdf(
 			pdf_path,
 			title=imp.import_title,
@@ -141,3 +143,8 @@ def rem_page_cb(page_no, total, method, adopted, base_c, new_c, metrics) -> None
 		frappe.db.commit()
 		log(import_name, "error", "parse", "Parse failed — see error on the import")
 		raise
+
+
+def delete_partial_documents(import_name: str) -> None:
+	for name in frappe.get_all("Source Document", filters={"import": import_name}, pluck="name"):
+		frappe.delete_doc("Source Document", name, ignore_permissions=True, force=True)

wikify/jobs/stalled.py

@@ -4,6 +4,7 @@
 from frappe.utils import add_to_date, now_datetime
 from frappe.utils.background_jobs import is_job_enqueued
 
+from wikify.engine import store
 from wikify.jobs._util import import_job_id, log, publish_progress
 
 RUNNING_STATUSES = ("Queued", "Parsing", "Remediating")
@@ -21,7 +22,11 @@ def fail_stalled_imports() -> None:
 	for row in imports:
 		if is_job_enqueued(import_job_id(row.name)):
 			continue
-		retry_hint = "Click Remediate to retry." if row.source_document else "Import the PDF again to retry."
+		retry_hint = (
+			"Click Resume import to continue."
+			if store.needs_resume(row.source_document)
+			else "Click Remediate to retry."
+		)
 		error = (
 			"The background job stopped before finishing (it timed out or the worker restarted). "
 			+ retry_hint

wikify/tests/test_imports_api.py

@@ -8,6 +8,8 @@
 from frappe.tests.utils import FrappeTestCase
 
 from wikify.api import imports as imports_api
+from wikify.engine import store
+from wikify.jobs import parse as parse_job
 from wikify.jobs._util import import_job_id
 from wikify.seed import seed_uncategorized_project
 
@@ -229,8 +231,8 @@ def test_remediation_refuses_while_the_import_job_is_running(self):
 
 
 class TestResumeImport(FrappeTestCase):
-	def make_import(self, status: str):
-		return frappe.get_doc(
+	def make_import(self, status: str, canonical_sources: tuple[str, ...] | None = None):
+		imp = frappe.get_doc(
 			{
 				"doctype": "Wikify Import",
 				"import_title": "Resume Test",
@@ -239,6 +241,20 @@ def make_import(self, status: str):
 				"status": status,
 			}
 		).insert(ignore_permissions=True)
+		if canonical_sources is not None:
+			source_document = store.create_document("Resume Test", import_name=imp.name)
+			for page_no, canonical_source in enumerate(canonical_sources, start=1):
+				frappe.get_doc(
+					{
+						"doctype": "Source Page",
+						"source_document": source_document,
+						"page_no": page_no,
+						"kind": "text",
+						"canonical_source": canonical_source,
+					}
+				).insert(ignore_permissions=True)
+			imp.db_set("source_document", source_document)
+		return imp
 
 	def test_a_stuck_import_requeues_its_parse_job(self):
 		for status in ("Parsing", "Remediating"):
@@ -297,3 +313,44 @@ def test_a_user_without_write_permission_cannot_resume(self):
 			enqueue.assert_not_called()
 		finally:
 			frappe.set_user("Administrator")
+
+	def test_a_failed_import_with_unremediated_pages_can_resume(self):
+		imp = self.make_import("Failed", canonical_sources=("vlm", "baseline", ""))
+		imp.db_set("error", "The background job stopped before finishing.")
+		with (
+			patch.object(imports_api, "is_job_enqueued", return_value=False),
+			patch.object(frappe, "enqueue") as enqueue,
+		):
+			self.assertTrue(imports_api.can_resume_import(imp.name))
+			imports_api.resume_import(imp.name)
+
+		enqueue.assert_called_once()
+		status, error = frappe.db.get_value("Wikify Import", imp.name, ["status", "error"])
+		self.assertEqual(status, "Queued")
+		self.assertFalse(error)
+
+	def test_an_import_with_every_page_remediated_cannot_resume(self):
+		for status in ("Remediating", "Failed"):
+			imp = self.make_import(status, canonical_sources=("vlm", "baseline", "cleanup"))
+			with (
+				self.subTest(status=status),
+				patch.object(imports_api, "is_job_enqueued", return_value=False),
+				patch.object(frappe, "enqueue") as enqueue,
+			):
+				self.assertFalse(imports_api.can_resume_import(imp.name))
+				with self.assertRaises(frappe.ValidationError):
+					imports_api.resume_import(imp.name)
+				enqueue.assert_not_called()
+
+	def test_a_failed_import_with_no_document_can_resume(self):
+		imp = self.make_import("Failed")
+		with patch.object(imports_api, "is_job_enqueued", return_value=False):
+			self.assertTrue(imports_api.can_resume_import(imp.name))
+
+	def test_reparse_deletes_the_half_parsed_document(self):
+		imp = self.make_import("Parsing")
+		half_parsed = store.create_document("Resume Test", import_name=imp.name)
+
+		parse_job.delete_partial_documents(imp.name)
+
+		self.assertFalse(frappe.db.exists("Source Document", half_parsed))

wikify/tests/test_stalled_imports.py

@@ -11,10 +11,19 @@
 
 
 class TestStalledImports(FrappeTestCase):
-	def make_import(self, status: str, minutes_since_modified: int):
+	def make_import(self, status: str, minutes_since_modified: int, unremediated_page: bool = False):
 		source_document = frappe.get_doc({"doctype": "Source Document", "title": "Stalled Test"}).insert(
 			ignore_permissions=True
 		)
+		if unremediated_page:
+			frappe.get_doc(
+				{
+					"doctype": "Source Page",
+					"source_document": source_document.name,
+					"page_no": 1,
+					"kind": "text",
+				}
+			).insert(ignore_permissions=True)
 		imp = frappe.get_doc(
 			{
 				"doctype": "Wikify Import",
@@ -52,6 +61,15 @@ def test_an_import_with_no_live_job_is_marked_failed(self):
 		self.assertIn("Click Remediate to retry", error)
 		self.assertTrue(frappe.db.exists("Import Log Entry", {"import": name, "level": "error"}))
 
+	def test_an_import_with_unremediated_pages_points_to_resume(self):
+		name = self.make_import("Remediating", 30, unremediated_page=True)
+
+		self.fail_stalled_imports(job_is_enqueued=False)
+
+		status, error = frappe.db.get_value("Wikify Import", name, ["status", "error"])
+		self.assertEqual(status, "Failed")
+		self.assertIn("Click Resume import to continue", error)
+
 	def test_an_import_with_a_live_job_is_left_running(self):
 		name = self.make_import("Remediating", 30)