Align Tally accounting storage with client storage policy

2026-08-20 14:25:57 +05:30
parent f23da90dec
commit 3912d2fd16
6 changed files with 204 additions and 47 deletions
@@ -1,2 +1,2 @@
-__version__ = "1.8.0"
+__version__ = "1.8.1"
 AGENT_NAME = "ERP Local Agent"
@@ -39,50 +39,184 @@ class LocalAccountingStore:
    """

    def __init__(self, storage_root: Path):
-        # Accounting databases are application data, not document-storage payloads.
-        # Keep them under the Local Agent data directory by default. An explicit
-        # ACCOUNTING_ROOT may override this without changing document STORAGE_ROOT.
-        configured = str(os.getenv("ACCOUNTING_ROOT", "") or "").strip()
-        install_root = Path(__file__).resolve().parents[1]
-        self.root = Path(configured).expanduser().resolve() if configured else (install_root / "data" / "accounting").resolve()
-        self.legacy_root = Path(storage_root).resolve() / "Accounting"
-        self.root.mkdir(parents=True, exist_ok=True)
+        # Accounting follows the same branch Storage Node root used by Permanent
+        # and Engagement storage. The ERP supplies the exact client-relative
+        # directory built with its shared client_folder_parts() policy.
+        self.storage_root = Path(storage_root).resolve()
+        self.storage_root.mkdir(parents=True, exist_ok=True)
+
+        install_root = Path(__file__).resolve().parents[1]
+        self.agent_data_root = (install_root / "data").resolve()
+        self.agent_data_root.mkdir(parents=True, exist_ok=True)
+        self.path_index_file = self.agent_data_root / "accounting_path_index.json"
+
+        # Historical locations retained only as migration sources.
+        self.legacy_flat_root = self.storage_root / "Accounting"
+        self.legacy_agent_root = self.agent_data_root / "accounting"
+
+        self._client_paths: dict[str, str] = self._load_path_index()
+
+    def _load_path_index(self) -> dict[str, str]:
+        if not self.path_index_file.exists():
+            return {}
+        try:
+            raw = json.loads(self.path_index_file.read_text(encoding="utf-8"))
+            if not isinstance(raw, dict):
+                return {}
+            return {str(k): str(v) for k, v in raw.items() if str(v).strip()}
+        except Exception:
+            return {}
+
+    def _save_path_index(self) -> None:
+        tmp = self.path_index_file.with_suffix(".json.tmp")
+        tmp.write_text(json.dumps(self._client_paths, indent=2, sort_keys=True), encoding="utf-8")
+        tmp.replace(self.path_index_file)
+
+    @staticmethod
+    def _safe_relative_dir(raw: str) -> Path:
+        text = str(raw or "").strip().replace("\\", "/")
+        if not text:
+            raise ValueError("Accounting storage path was not supplied by ERP.")
+        path = Path(text)
+        if path.is_absolute():
+            raise ValueError("Accounting storage path must be relative to the configured Storage Node root.")
+        safe_parts = []
+        for part in path.parts:
+            value = str(part or "").strip()
+            if value in {"", ".", ".."}:
+                raise ValueError("Unsafe Accounting storage path received from ERP.")
+            if "/" in value or "\\" in value:
+                raise ValueError("Unsafe Accounting storage segment received from ERP.")
+            safe_parts.append(value)
+        relative = Path(*safe_parts)
+        # Accounting paths are intentionally constrained to Accounting/Clients/...
+        parts_lower = [p.lower() for p in relative.parts]
+        if len(parts_lower) < 4 or parts_lower[0] != "accounting" or parts_lower[1] != "clients":
+            raise ValueError("Accounting storage path does not follow the ERP Storage Node client policy.")
+        return relative
+
+    def bind_client_path(self, client_id: int, accounting_relative_dir: str) -> Path:
+        relative = self._safe_relative_dir(accounting_relative_dir)
+        target_dir = (self.storage_root / relative).resolve()
+        if os.path.commonpath([str(self.storage_root), str(target_dir)]) != str(self.storage_root):
+            raise ValueError("Accounting storage path escapes the configured Storage Node root.")

-    def _migrate_legacy_client_if_needed(self, client_id: int) -> None:
        key = self._client_key(client_id)
-        new_dir = self.root / key
-        new_db = new_dir / f"{key}.act"
-        old_dir = self.legacy_root / key
-        old_db = old_dir / f"{key}.act"
-        if new_db.exists() or not old_db.exists():
+        target_dir.mkdir(parents=True, exist_ok=True)
+        target_db = target_dir / f"{key}.act"
+
+        self._client_paths[str(int(client_id))] = relative.as_posix()
+        self._save_path_index()
+        self._migrate_existing_database(client_id, target_db)
+        return target_db
+
+    @staticmethod
+    def _database_activity_mtime(path: Path) -> float:
+        stamps = []
+        for candidate in (path, Path(str(path) + "-wal"), Path(str(path) + "-shm")):
+            try:
+                if candidate.exists():
+                    stamps.append(candidate.stat().st_mtime)
+            except OSError:
+                pass
+        return max(stamps) if stamps else 0.0
+
+    @staticmethod
+    def _sqlite_backup(source: Path, destination: Path) -> None:
+        destination.parent.mkdir(parents=True, exist_ok=True)
+        tmp = destination.with_suffix(destination.suffix + ".migrating")
+        if tmp.exists():
+            tmp.unlink()
+        src = sqlite3.connect(source, timeout=60)
+        dst = sqlite3.connect(tmp, timeout=60)
+        try:
+            src.execute("PRAGMA busy_timeout=60000")
+            dst.execute("PRAGMA busy_timeout=60000")
+            src.backup(dst)
+            dst.commit()
+        finally:
+            dst.close()
+            src.close()
+        tmp.replace(destination)
+
+    def _migration_candidates(self, client_id: int, target_db: Path) -> list[Path]:
+        key = self._client_key(client_id)
+        filename = f"{key}.act"
+        candidates: list[Path] = []
+
+        # Original Phase 1-7 flat Storage Root location.
+        candidates.append(self.legacy_flat_root / key / filename)
+
+        # Temporary v1.8.0 Local-Agent data location.
+        candidates.append(self.legacy_agent_root / key / filename)
+
+        # If a client was renamed, an earlier canonical client-folder path may exist.
+        accounting_clients = self.storage_root / "Accounting" / "Clients"
+        if accounting_clients.exists():
+            for path in accounting_clients.glob(f"*/*/{filename}"):
+                candidates.append(path)
+
+        unique = []
+        seen = set()
+        for path in candidates:
+            try:
+                resolved = path.resolve()
+            except Exception:
+                resolved = path
+            if resolved == target_db.resolve():
+                continue
+            marker = str(resolved).lower()
+            if marker in seen or not path.exists():
+                continue
+            seen.add(marker)
+            unique.append(path)
+        return unique
+
+    def _migrate_existing_database(self, client_id: int, target_db: Path) -> None:
+        if target_db.exists():
            return
-        new_dir.mkdir(parents=True, exist_ok=True)
-        # Copy, do not delete, the legacy database. The old copy remains a recovery
-        # fallback while all new reads/writes switch to Local Agent data/accounting.
-        shutil.copy2(old_db, new_db)
-        for suffix in ("-wal", "-shm"):
-            src = Path(str(old_db) + suffix)
-            if src.exists():
-                shutil.copy2(src, Path(str(new_db) + suffix))
+        candidates = self._migration_candidates(client_id, target_db)
+        if not candidates:
+            return
+        source = max(candidates, key=self._database_activity_mtime)
+        self._sqlite_backup(source, target_db)
+        # Source is deliberately left untouched as a recovery copy.
+
+    def _bound_relative_dir(self, client_id: int) -> Path | None:
+        raw = self._client_paths.get(str(int(client_id)))
+        if not raw:
+            return None
+        try:
+            return self._safe_relative_dir(raw)
+        except Exception:
+            return None

    @staticmethod
    def _client_key(client_id: int) -> str:
        return f"client_{int(client_id):08d}"

    def client_dir(self, client_id: int) -> Path:
-        return self.root / self._client_key(client_id)
+        relative = self._bound_relative_dir(client_id)
+        if relative is None:
+            # Do not silently create a new Accounting DB in a non-policy location.
+            # Existing legacy DBs remain discoverable only after ERP supplies its
+            # canonical client path on the next status/initialize command.
+            return self.storage_root / "Accounting" / "Clients" / "_UNBOUND" / self._client_key(client_id)
+        return self.storage_root / relative

    def db_path(self, client_id: int) -> Path:
-        self._migrate_legacy_client_if_needed(client_id)
        key = self._client_key(client_id)
        return self.client_dir(client_id) / f"{key}.act"

    def exists(self, client_id: int) -> bool:
-        self._migrate_legacy_client_if_needed(client_id)
        return self.db_path(client_id).is_file()

    def connect(self, client_id: int):
-        self._migrate_legacy_client_if_needed(client_id)
+        if self._bound_relative_dir(client_id) is None:
+            raise RuntimeError(
+                "Accounting storage path is not bound to the ERP client folder policy. "
+                "Refresh Tools → Tally so ERP can supply the canonical Storage Node path."
+            )
        path = self.db_path(client_id)
        path.parent.mkdir(parents=True, exist_ok=True)
        db = sqlite3.connect(path, timeout=60)
@@ -19,6 +19,7 @@ class AgentCommandProcessor:
        command_id = str(command.get("command_id") or "").strip()
        action = str(command.get("action") or "").strip()
        payload = command.get("payload") or {}
+        self._bind_accounting_storage(payload)
        result: dict[str, Any] | None = None
        error: str | None = None
        ok = False
@@ -53,6 +54,13 @@ class AgentCommandProcessor:
            self.logger.exception("Agent command failed action=%s command_id=%s: %s", action, command_id, exc)
        return {"type": "command_result", "command_id": command_id, "ok": ok, "result": result, "error": error, "agent_time_utc": datetime.now(timezone.utc).isoformat()}

+    def _bind_accounting_storage(self, payload: dict[str, Any]) -> None:
+        client_id = payload.get("client_id")
+        relative_dir = str(payload.get("accounting_relative_dir") or "").strip()
+        if client_id in (None, "") or not relative_dir:
+            return
+        self.store.bind_client_path(int(client_id), relative_dir)
+
    def _agent_info(self) -> dict[str, Any]:
        return {
            "name": "ERP Local Agent", "version": __version__,