From 2331471a05050a3c80b2de86e9c891fc0432baee Mon Sep 17 00:00:00 2001 From: vishwajeet-13 Date: Sat, 10 Oct 2026 08:01:49 +0530 Subject: [PATCH] fix(import): resume failed imports and skip finished or half-parsed ones - Failed imports with unfinished pages (stall check, job timeout) can be resumed. - Resume is no longer offered when every page is already remediated. - Re-parsing deletes the half-parsed Source Document first. - The stalled error points to Resume when the import can continue. --- frontend/src/pages/ImportDetail.vue | 16 +++++--- wikify/api/imports.py | 8 +++- wikify/engine/store.py | 8 ++++ wikify/jobs/parse.py | 7 ++++ wikify/jobs/stalled.py | 7 +++- wikify/tests/test_imports_api.py | 61 +++++++++++++++++++++++++++- wikify/tests/test_stalled_imports.py | 20 ++++++++- 7 files changed, 115 insertions(+), 12 deletions(-) diff --git a/frontend/src/pages/ImportDetail.vue b/frontend/src/pages/ImportDetail.vue index 0351905..42952a8 100644 --- a/frontend/src/pages/ImportDetail.vue +++ b/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 diff --git a/wikify/api/imports.py b/wikify/api/imports.py index 6a2aea5..ae1d023 100644 --- a/wikify/api/imports.py +++ b/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 diff --git a/wikify/engine/store.py b/wikify/engine/store.py index 0300ff8..400d766 100644 --- a/wikify/engine/store.py +++ b/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) diff --git a/wikify/jobs/parse.py b/wikify/jobs/parse.py index a836a90..f5ea5b6 100644 --- a/wikify/jobs/parse.py +++ b/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) diff --git a/wikify/jobs/stalled.py b/wikify/jobs/stalled.py index 5037e41..2197379 100644 --- a/wikify/jobs/stalled.py +++ b/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 diff --git a/wikify/tests/test_imports_api.py b/wikify/tests/test_imports_api.py index 025c6ba..6872961 100644 --- a/wikify/tests/test_imports_api.py +++ b/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)) diff --git a/wikify/tests/test_stalled_imports.py b/wikify/tests/test_stalled_imports.py index 13d3da4..4a4c1b7 100644 --- a/wikify/tests/test_stalled_imports.py +++ b/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)