Add async SQLite job manager for cash payment extraction

app/modules/accounting/cash_payment_ui.py

@@ -307,6 +307,52 @@ async def start_cash_payment_cache(request: Request):
         db.close()
 
 
+@router.post("/cache/cancel")
+async def cancel_cash_payment_cache(request: Request):
+    form = await request.form()
+    validate_csrf(request, str(form.get("csrf_token") or ""))
+    db = CommonSessionLocal()
+    try:
+        user, response = _require_partner(request, db, "accounting.tally.view")
+        if response:
+            return JSONResponse({"ok": False, "error": "Access denied."}, status_code=403)
+        clients, scope = _visible_clients(db, request, user)
+        try:
+            client_id = int(form.get("client_id") or 0)
+        except Exception:
+            client_id = 0
+        selected_client = next((row for row in clients if int(row.id) == client_id), None)
+        if not selected_client:
+            return JSONResponse({"ok": False, "error": "Select a valid client."}, status_code=400)
+        job_id = str(form.get("job_id") or "").strip()
+        date_from = str(form.get("date_from") or "").strip()
+        if not job_id:
+            return JSONResponse({"ok": False, "error": "Extraction job id is required."}, status_code=400)
+        try:
+            parsed_from = date.fromisoformat(date_from)
+        except Exception:
+            return JSONResponse({"ok": False, "error": "Valid extraction From date is required."}, status_code=400)
+        node = get_active_storage_node_for_branch(db, scope.tenant_id, scope.branch_id)
+        if not node or not _node_online(node):
+            return JSONResponse({"ok": False, "error": "ERP Local Agent is offline."}, status_code=409)
+        result = request_agent_command(
+            node.node_code,
+            "accounting_cash_payment_cache_cancel",
+            {
+                **_accounting_storage_payload(selected_client, _financial_year_for_date(parsed_from)),
+                "job_id": job_id,
+            },
+            timeout_seconds=15,
+        )
+        if not result.get("ok"):
+            return JSONResponse({"ok": False, "error": str(result.get("error") or "Could not cancel extraction.")}, status_code=409)
+        return JSONResponse({"ok": True, **(result.get("result") or {})})
+    except Exception as exc:
+        return JSONResponse({"ok": False, "error": str(exc)}, status_code=502)
+    finally:
+        db.close()
+
+
 @router.get("/cache/progress")
 def cash_payment_cache_progress(request: Request, client_id: int, job_id: str, date_from: str):
     db = CommonSessionLocal()

app/modules/accounting/templates/accounting/cash_payment_review.html

@@ -152,7 +152,10 @@
     </div>
     <p class="mt-4 text-xs text-slate-500">Only one day is requested from TallyPrime at a time. The agent waits about 3 seconds after every successful day. If Tally responds slowly, it cools for 20 seconds and retries only once. The tunnel heartbeat stays responsive because progress is read from SQLite, not from Tally.</p>
     <div id="cash-progress-error" class="mt-4 hidden rounded-xl border border-amber-200 bg-amber-50 p-3 text-sm text-amber-900"></div>
-    <div class="mt-4 flex justify-end"><button type="button" id="cash-progress-close" class="hidden rounded-xl border border-slate-300 bg-white px-4 py-2 text-sm font-semibold text-slate-700">Close</button></div>
+    <div class="mt-4 flex justify-end gap-2">
+      <button type="button" id="cash-progress-cancel" class="hidden rounded-xl border border-amber-300 bg-amber-50 px-4 py-2 text-sm font-semibold text-amber-900">Cancel Extraction</button>
+      <button type="button" id="cash-progress-close" class="hidden rounded-xl border border-slate-300 bg-white px-4 py-2 text-sm font-semibold text-slate-700">Close</button>
+    </div>
   </div>
 </div>
 <script>
@@ -170,13 +173,18 @@
   const currentDate = document.getElementById('cash-progress-date');
   const errorBox = document.getElementById('cash-progress-error');
   const closeButton = document.getElementById('cash-progress-close');
+  const cancelButton = document.getElementById('cash-progress-cancel');
   let pollTimer = null;
+  let activeJobId = '';
+  let activeClientId = '';
+  let activeDateFrom = '';
 
   function show() {
     overlay.classList.remove('hidden');
     overlay.classList.add('flex');
     errorBox.classList.add('hidden');
     closeButton.classList.add('hidden');
+    cancelButton.classList.add('hidden');
   }
   function stopPolling() {
     if (pollTimer) window.clearTimeout(pollTimer);
@@ -189,6 +197,7 @@
     errorBox.textContent = message || 'The extraction did not complete.';
     errorBox.classList.remove('hidden');
     closeButton.classList.remove('hidden');
+    cancelButton.classList.add('hidden');
   }
   function update(job) {
     const value = Math.max(0, Math.min(100, Number(job.percent || 0)));
@@ -199,6 +208,7 @@
     vouchers.textContent = String(job.vouchers_cached || 0);
     requests.textContent = String(job.tally_requests || 0);
     currentDate.textContent = job.current_date || '—';
+    if (job.can_cancel) cancelButton.classList.remove('hidden'); else cancelButton.classList.add('hidden');
   }
   async function poll(jobId, clientId, dateFrom, tallyGuid, dateTo) {
     try {
@@ -224,6 +234,10 @@
         window.location.assign(target.toString());
         return;
       }
+      if (job.status === 'cancelled') {
+        fail('Extraction was cancelled safely. Already cached days remain available and can be resumed later.');
+        return;
+      }
       if (job.status === 'paused' || job.status === 'failed') {
         fail(job.error || 'Tally extraction paused. Restart/verify TallyPrime and click Extract / Refresh & Analyze again to resume.');
         return;
@@ -245,12 +259,36 @@
       const data = await response.json();
       if (!response.ok || !data.ok) throw new Error(data.error || 'Could not start extraction.');
       const job = data.job || {};
+      activeJobId = job.job_id || '';
+      activeClientId = String(fd.get('client_id') || '');
+      activeDateFrom = String(fd.get('date_from') || '');
       update(job);
       poll(job.job_id, fd.get('client_id'), fd.get('date_from'), fd.get('tally_guid'), fd.get('date_to'));
     } catch (err) {
       fail(err && err.message ? err.message : String(err));
     }
   });
+  cancelButton.addEventListener('click', async function () {
+    if (!activeJobId || !activeClientId) return;
+    cancelButton.disabled = true;
+    text.textContent = 'Requesting a safe stop after the current Tally request…';
+    try {
+      const fd = new FormData();
+      fd.set('csrf_token', document.getElementById('cash-csrf-token').value);
+      fd.set('client_id', activeClientId);
+      fd.set('job_id', activeJobId);
+      fd.set('date_from', activeDateFrom);
+      const response = await fetch('/tools/accounting/cash-payments/cache/cancel', {method: 'POST', body: fd, headers: {'Accept': 'application/json'}});
+      const data = await response.json();
+      if (!response.ok || !data.ok) throw new Error(data.error || 'Could not cancel extraction.');
+      if (data.job) update(data.job);
+      cancelButton.classList.add('hidden');
+    } catch (err) {
+      errorBox.textContent = err && err.message ? err.message : String(err);
+      errorBox.classList.remove('hidden');
+      cancelButton.disabled = false;
+    }
+  });
   closeButton.addEventListener('click', function () {
     overlay.classList.add('hidden'); overlay.classList.remove('flex');
   });

app/modules/documents/agent_package.py

@@ -4,7 +4,7 @@ import io
 from pathlib import Path
 import zipfile
 
-ERP_LOCAL_AGENT_VERSION = "1.22.8"
+ERP_LOCAL_AGENT_VERSION = "1.22.9"
 ERP_LOCAL_AGENT_NAME = "ERP Local Agent"
 RUNTIME_ROOT = Path(__file__).resolve().parent / "local_agent_runtime"
 _DETERMINISTIC_ZIP_TIMESTAMP = (2026, 1, 1, 0, 0, 0)

app/modules/documents/local_agent_runtime/erp_local_agent/__init__.py

@@ -1,2 +1,2 @@
-__version__ = "1.22.8"
+__version__ = "1.22.9"
 AGENT_NAME = "ERP Local Agent"

app/modules/documents/local_agent_runtime/erp_local_agent/commands.py

@@ -18,6 +18,7 @@ _CASH_TALLY_EXTRACTION_LOCK = threading.Lock()
 _CASH_CACHE_BUSY = threading.Event()
 _CASH_CACHE_THREADS: dict[str, threading.Thread] = {}
 _CASH_CACHE_THREADS_LOCK = threading.Lock()
+_CASH_CACHE_CANCEL_REQUESTS: set[str] = set()
 
 def is_tally_cache_busy() -> bool:
     return _CASH_CACHE_BUSY.is_set()
@@ -73,6 +74,8 @@ class AgentCommandProcessor:
                 result = self._cash_payment_cache_start(payload)
             elif action == "accounting_cash_payment_cache_status":
                 result = self._cash_payment_cache_status(payload)
+            elif action == "accounting_cash_payment_cache_cancel":
+                result = self._cash_payment_cache_cancel(payload)
             elif action == "accounting_cash_payment_cache_analyze":
                 result = self._cash_payment_cache_analyze(payload)
             elif action == "accounting_cash_payment_compliance":
@@ -768,10 +771,15 @@ class AgentCommandProcessor:
                     return {"job": self._cash_payment_cache_status({"client_id": client_id, "job_id": prior["job_id"]})["job"], "resumed": False, "already_running": True}
                 raise ValueError("Another Tally extraction is already running on this workstation. Please wait for it to finish.")
 
-        company, company_name = self._resolve_open_company(payload)
-        if str(company.get("guid") or "").strip() != tally_guid:
-            raise ValueError("The selected Tally company is not the currently resolved company.")
+        # IMPORTANT: starting a cache job must never call TallyPrime.  The UI/ERP
+        # command must receive a job id immediately, while all Tally I/O happens
+        # only inside the background worker.  Resolve the mapped company name
+        # from the local .act SQLite store; the worker validates the actually
+        # open Tally company after the command result has already been returned.
         mapping = self.store.get_active_mapping_by_guid(client_id, tally_guid)
+        company_name = str(mapping.get("company_name") or "").strip()
+        if not company_name:
+            raise ValueError("The mapped Tally company name is not available in local accounting storage. Refresh the Tally mapping first.")
         pause_seconds = max(2.0, min(float(payload.get("tally_pause_seconds") or 3.0), 15.0))
         settings = {
             "cash_limit": float(payload.get("cash_limit") or 10000.0),
@@ -810,9 +818,12 @@ class AgentCommandProcessor:
 
         worker_payload = dict(payload)
         worker_payload["tally_pause_seconds"] = pause_seconds
+        worker_payload["company_name"] = company_name
+        with _CASH_CACHE_THREADS_LOCK:
+            _CASH_CACHE_CANCEL_REQUESTS.discard(job_id)
         thread = threading.Thread(
             target=self._run_cash_payment_cache_job,
-            args=(job_id, worker_payload, dict(company), company_name, dict(mapping)),
+            args=(job_id, worker_payload, dict(mapping)),
             name=f"tally-cash-cache-{job_id[:8]}",
             daemon=True,
         )
@@ -821,13 +832,25 @@ class AgentCommandProcessor:
         thread.start()
         return {"job": self._cash_payment_cache_status({"client_id": client_id, "job_id": job_id})["job"], "resumed": resume, "already_running": False}
 
-    def _run_cash_payment_cache_job(self, job_id: str, payload: dict[str, Any], company: dict[str, Any], company_name: str, mapping: dict[str, Any]) -> None:
+    def _run_cash_payment_cache_job(self, job_id: str, payload: dict[str, Any], mapping: dict[str, Any]) -> None:
         client_id = int(payload.get("client_id"))
         pause_seconds = max(2.0, min(float(payload.get("tally_pause_seconds") or 3.0), 15.0))
         requested_by = payload.get("requested_by_user_id")
         _CASH_TALLY_EXTRACTION_LOCK.acquire()
         _CASH_CACHE_BUSY.set()
         try:
+            self._cash_cache_update(client_id, job_id, status="running", stage="Connecting to TallyPrime in background", current_date="")
+            if job_id in _CASH_CACHE_CANCEL_REQUESTS:
+                self._cash_cache_update(client_id, job_id, status="cancelled", stage="Cancelled before Tally connection", completed_at_utc=datetime.now(timezone.utc).isoformat())
+                return
+
+            # This is deliberately inside the worker.  It may be slow, but it can
+            # no longer hold open the ERP start-job command or block its timeout.
+            company, company_name = self._resolve_open_company(payload)
+            expected_guid = str(payload.get("tally_guid") or "").strip()
+            if str(company.get("guid") or "").strip() != expected_guid:
+                raise ValueError("The selected Tally company is not the currently resolved company.")
+
             self._cash_cache_update(client_id, job_id, status="running", stage="Reading cash ledger masters", current_date="")
             # One master request only, then a generous cooling gap before vouchers.
             masters = None
@@ -862,6 +885,9 @@ class AgentCommandProcessor:
                     (job_id,),
                 ).fetchall()
             for day_row in pending:
+                if job_id in _CASH_CACHE_CANCEL_REQUESTS:
+                    self._cash_cache_update(client_id, job_id, status="cancelled", stage="Cancelled safely", current_date="", completed_at_utc=datetime.now(timezone.utc).isoformat())
+                    return
                 day = str(day_row[0])
                 self._cash_cache_update(client_id, job_id, status="running", stage="Extracting one day from TallyPrime", current_date=day)
                 vouchers = None
@@ -917,6 +943,21 @@ class AgentCommandProcessor:
             _CASH_TALLY_EXTRACTION_LOCK.release()
             with _CASH_CACHE_THREADS_LOCK:
                 _CASH_CACHE_THREADS.pop(job_id, None)
+                _CASH_CACHE_CANCEL_REQUESTS.discard(job_id)
+
+    def _cash_payment_cache_cancel(self, payload: dict[str, Any]) -> dict[str, Any]:
+        client_id = int(payload.get("client_id"))
+        job_id = str(payload.get("job_id") or "").strip()
+        row = self._cash_cache_job_row(client_id, job_id)
+        if not row:
+            raise ValueError("Cash payment cache job was not found on this workstation.")
+        status = str(row.get("status") or "")
+        if status in {"completed", "cancelled", "failed", "paused"}:
+            return {"job": self._cash_payment_cache_status({"client_id": client_id, "job_id": job_id})["job"], "cancel_requested": False}
+        with _CASH_CACHE_THREADS_LOCK:
+            _CASH_CACHE_CANCEL_REQUESTS.add(job_id)
+        self._cash_cache_update(client_id, job_id, stage="Cancellation requested; stopping after the current safe Tally request")
+        return {"job": self._cash_payment_cache_status({"client_id": client_id, "job_id": job_id})["job"], "cancel_requested": True}
 
     def _cash_payment_cache_status(self, payload: dict[str, Any]) -> dict[str, Any]:
         client_id = int(payload.get("client_id"))
@@ -948,7 +989,8 @@ class AgentCommandProcessor:
                 "completed_at_utc": row.get("completed_at_utc"),
                 "background": True,
                 "sqlite_cache": True,
-                "can_resume": str(row.get("status") or "") in {"paused", "failed", "interrupted"},
+                "can_resume": str(row.get("status") or "") in {"paused", "failed", "interrupted", "cancelled"},
+                "can_cancel": str(row.get("status") or "") in {"queued", "running", "cooling"},
             },
             "agent": self._agent_info(),
         }