diff --git a/fusion_accounting_reports/__manifest__.py b/fusion_accounting_reports/__manifest__.py
index 87945d79..da493654 100644
--- a/fusion_accounting_reports/__manifest__.py
+++ b/fusion_accounting_reports/__manifest__.py
@@ -1,6 +1,6 @@
 {
     'name': 'Fusion Accounting Reports',
-    'version': '19.0.1.0.3',
+    'version': '19.0.1.0.4',
     'category': 'Accounting/Accounting',
     'summary': 'AI-augmented financial reports (P&L, balance sheet, trial balance, GL).',
     'description': """
diff --git a/fusion_accounting_reports/models/__init__.py b/fusion_accounting_reports/models/__init__.py
index f8cf3dce..4a3eb8fd 100644
--- a/fusion_accounting_reports/models/__init__.py
+++ b/fusion_accounting_reports/models/__init__.py
@@ -1 +1,2 @@
 from . import fusion_report
+from . import fusion_report_engine
diff --git a/fusion_accounting_reports/models/fusion_report_engine.py b/fusion_accounting_reports/models/fusion_report_engine.py
new file mode 100644
index 00000000..4f030f98
--- /dev/null
+++ b/fusion_accounting_reports/models/fusion_report_engine.py
@@ -0,0 +1,245 @@
+"""The reports engine - orchestrator for all report computation.
+
+5-method public API. All controllers, AI tools, wizards, exports must
+go through these methods; no direct ORM aggregation queries from
+anywhere else.
+
+Internal pipeline (per report run):
+1. Validate (period valid, company allowed, report exists)
+2. Fetch account hierarchy (cached per (company, fiscal_year))
+3. Aggregate move lines per account (the SQL workhorse)
+4. Resolve line_specs into report rows
+5. (Optional) Compute comparison-period rows
+6. (Optional) Detect anomalies (deferred to later tasks)
+"""
+
+import logging
+from datetime import date
+
+from odoo import _, api, models
+from odoo.exceptions import ValidationError
+
+from ..services.account_hierarchy import build_tree
+from ..services.date_periods import Period, comparison_period as _comp_period
+from ..services.drill_down_resolver import fetch_drill_down
+from ..services.line_resolver import resolve as _resolve_lines
+from ..services.totaling import TotalLine
+
+_logger = logging.getLogger(__name__)
+
+
+class FusionReportEngine(models.AbstractModel):
+    _name = "fusion.report.engine"
+    _description = "Fusion Financial Reports Engine"
+
+    # ============================================================
+    # PUBLIC API (5 methods)
+    # ============================================================
+
+    @api.model
+    def compute_pnl(
+        self, period: Period, *, comparison: str = 'none',
+        company_id: int | None = None,
+    ) -> dict:
+        """Income statement (P&L) for the given period."""
+        report = self._get_report('pnl', company_id=company_id)
+        return self._compute(
+            report, period, comparison=comparison, company_id=company_id,
+        )
+
+    @api.model
+    def compute_balance_sheet(
+        self, date_to: date, *, comparison: str = 'none',
+        company_id: int | None = None,
+    ) -> dict:
+        """Balance sheet AS OF date_to. Period.date_from is set to a
+        far-past date so balances are cumulative-since-inception."""
+        report = self._get_report('balance_sheet', company_id=company_id)
+        period = Period(
+            date_from=date(1970, 1, 1),
+            date_to=date_to,
+            label=f"As of {date_to}",
+        )
+        return self._compute(
+            report, period, comparison=comparison, company_id=company_id,
+        )
+
+    @api.model
+    def compute_trial_balance(
+        self, period: Period, *, company_id: int | None = None,
+    ) -> dict:
+        """Trial balance for the given period - every account with
+        non-zero balance."""
+        report = self._get_report('trial_balance', company_id=company_id)
+        return self._compute(
+            report, period, comparison='none', company_id=company_id,
+        )
+
+    @api.model
+    def compute_gl(
+        self, period: Period, *, account_ids: list | None = None,
+        company_id: int | None = None,
+    ) -> dict:
+        """General ledger for the given period.
+
+        Returns per-account move-line listings rather than aggregated rows."""
+        report = self._get_report('general_ledger', company_id=company_id)
+        company_id = company_id or self.env.company.id
+        result = self._compute(
+            report, period, comparison='none', company_id=company_id,
+        )
+        gl_by_account = {}
+        target_ids = account_ids or list(result.get('account_totals', {}).keys())
+        for acct_id in target_ids:
+            gl_by_account[acct_id] = fetch_drill_down(
+                self.env,
+                account_id=acct_id,
+                date_from=period.date_from,
+                date_to=period.date_to,
+                company_id=company_id,
+                limit=200,
+            )
+        result['gl_by_account'] = gl_by_account
+        return result
+
+    @api.model
+    def drill_down(
+        self, *, account_id: int, period: Period,
+        company_id: int | None = None,
+    ) -> list:
+        """Drill into a report line: list the journal items behind it."""
+        company_id = company_id or self.env.company.id
+        return fetch_drill_down(
+            self.env,
+            account_id=account_id,
+            date_from=period.date_from,
+            date_to=period.date_to,
+            company_id=company_id,
+            limit=500,
+        )
+
+    # ============================================================
+    # PRIVATE HELPERS
+    # ============================================================
+
+    def _get_report(self, report_type: str, *, company_id: int | None = None):
+        """Look up the active fusion.report definition for a given
+        type+company. If no per-company override, falls back to global
+        (company_id=False)."""
+        Report = self.env['fusion.report'].sudo()
+        company_id = company_id or self.env.company.id
+        report = Report.search(
+            [
+                ('report_type', '=', report_type),
+                ('active', '=', True),
+                '|',
+                ('company_id', '=', company_id),
+                ('company_id', '=', False),
+            ],
+            order='company_id desc nulls last',
+            limit=1,
+        )
+        if not report:
+            raise ValidationError(
+                _("No active fusion.report definition for type '%s'") % report_type
+            )
+        return report
+
+    def _fetch_accounts(self, company_id):
+        """Fetch all accounts for a company, return flat dict + tree."""
+        Account = self.env['account.account'].sudo()
+        records = Account.search([('company_ids', 'in', company_id)])
+        # account.account doesn't carry a parent_id in V19 - we use
+        # account_type prefixes instead, so parent_id is always None here.
+        flat = [
+            {
+                'id': a.id,
+                'code': a.code,
+                'name': a.name,
+                'account_type': a.account_type or '',
+                'parent_id': None,
+            }
+            for a in records
+        ]
+        accounts_by_id = {a['id']: a for a in flat}
+        tree = build_tree(flat)
+        return accounts_by_id, tree
+
+    def _aggregate_period(self, period: Period, company_id: int) -> dict:
+        """SQL aggregate per account_id for a period.
+
+        Raw SQL for performance; this is the perf-critical step."""
+        self.env.cr.execute(
+            """
+            SELECT account_id,
+                   COALESCE(SUM(debit), 0) AS d,
+                   COALESCE(SUM(credit), 0) AS c,
+                   COALESCE(SUM(balance), 0) AS b
+            FROM account_move_line
+            WHERE parent_state = 'posted'
+              AND company_id = %s
+              AND date >= %s
+              AND date <= %s
+            GROUP BY account_id
+            """,
+            (company_id, period.date_from, period.date_to),
+        )
+        out = {}
+        for row in self.env.cr.fetchall():
+            out[row[0]] = TotalLine(
+                debit=float(row[1] or 0),
+                credit=float(row[2] or 0),
+                balance=float(row[3] or 0),
+            )
+        return out
+
+    def _compute(
+        self, report, period: Period, *, comparison: str,
+        company_id: int | None = None,
+    ) -> dict:
+        """Shared computation pipeline. Returns dict with rows, totals,
+        metadata."""
+        company_id = company_id or self.env.company.id
+
+        accounts_by_id, _tree = self._fetch_accounts(company_id)
+
+        account_totals = self._aggregate_period(period, company_id)
+
+        comp_totals = None
+        comp_period = None
+        if comparison and comparison != 'none':
+            comp_period = _comp_period(period, comparison)
+            if comp_period:
+                comp_totals = self._aggregate_period(comp_period, company_id)
+
+        rows = _resolve_lines(
+            report.line_specs or [],
+            account_totals=account_totals,
+            accounts_by_id=accounts_by_id,
+            comparison_totals=comp_totals,
+        )
+
+        return {
+            'report_id': report.id,
+            'report_name': report.name,
+            'report_type': report.report_type,
+            'period': {
+                'date_from': str(period.date_from),
+                'date_to': str(period.date_to),
+                'label': period.label,
+            },
+            'comparison_period': (
+                {
+                    'date_from': str(comp_period.date_from),
+                    'date_to': str(comp_period.date_to),
+                    'label': comp_period.label,
+                }
+                if comp_period
+                else None
+            ),
+            'company_id': company_id,
+            'rows': rows,
+            'account_totals': {
+                aid: tl.balance for aid, tl in account_totals.items()
+            },
+        }
diff --git a/fusion_accounting_reports/tests/__init__.py b/fusion_accounting_reports/tests/__init__.py
index 14bd9143..b2488e98 100644
--- a/fusion_accounting_reports/tests/__init__.py
+++ b/fusion_accounting_reports/tests/__init__.py
@@ -3,3 +3,4 @@ from . import test_currency_conversion
 from . import test_fusion_report
 from . import test_line_resolver
 from . import test_drill_down_resolver
+from . import test_fusion_report_engine
diff --git a/fusion_accounting_reports/tests/test_fusion_report_engine.py b/fusion_accounting_reports/tests/test_fusion_report_engine.py
new file mode 100644
index 00000000..43e4e21a
--- /dev/null
+++ b/fusion_accounting_reports/tests/test_fusion_report_engine.py
@@ -0,0 +1,109 @@
+"""Tests for fusion.report.engine AbstractModel."""
+
+from datetime import date
+
+from odoo.exceptions import ValidationError
+from odoo.tests.common import TransactionCase, tagged
+from odoo.addons.fusion_accounting_reports.services.date_periods import Period
+
+
+@tagged('post_install', '-at_install')
+class TestFusionReportEngine(TransactionCase):
+
+    def setUp(self):
+        super().setUp()
+        self.pnl_report = self.env['fusion.report'].create({
+            'name': 'Test P&L Engine',
+            'code': 'test_pnl_engine',
+            'report_type': 'pnl',
+            'line_specs': [
+                {'label': 'Revenue', 'account_type_prefix': 'income_', 'sign': 1},
+                {'label': 'Expenses', 'account_type_prefix': 'expense_', 'sign': -1},
+                {'label': 'Net Profit', 'compute': 'subtotal', 'above': 2},
+            ],
+            'company_id': self.env.company.id,
+        })
+
+    def test_engine_model_exists(self):
+        self.assertIn('fusion.report.engine', self.env.registry)
+
+    def test_compute_pnl_returns_dict_with_rows(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_pnl(
+            period, company_id=self.env.company.id,
+        )
+        self.assertIn('rows', result)
+        self.assertIn('report_type', result)
+        self.assertEqual(result['report_type'], 'pnl')
+
+    def test_compute_balance_sheet(self):
+        self.env['fusion.report'].create({
+            'name': 'Test BS',
+            'code': 'test_bs_engine',
+            'report_type': 'balance_sheet',
+            'line_specs': [
+                {'label': 'Assets', 'account_type_prefix': 'asset_', 'sign': 1},
+            ],
+            'company_id': self.env.company.id,
+        })
+        result = self.env['fusion.report.engine'].compute_balance_sheet(
+            date(2026, 4, 19), company_id=self.env.company.id,
+        )
+        self.assertEqual(result['report_type'], 'balance_sheet')
+        self.assertEqual(result['period']['date_to'], '2026-04-19')
+
+    def test_compute_trial_balance(self):
+        self.env['fusion.report'].create({
+            'name': 'Test TB',
+            'code': 'test_tb_engine',
+            'report_type': 'trial_balance',
+            'line_specs': [],
+            'company_id': self.env.company.id,
+        })
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_trial_balance(
+            period, company_id=self.env.company.id,
+        )
+        self.assertEqual(result['report_type'], 'trial_balance')
+
+    def test_compute_pnl_with_comparison(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        result = self.env['fusion.report.engine'].compute_pnl(
+            period,
+            comparison='previous_year',
+            company_id=self.env.company.id,
+        )
+        self.assertIsNotNone(result.get('comparison_period'))
+        self.assertEqual(result['comparison_period']['date_to'], '2025-12-31')
+
+    def test_drill_down_returns_list(self):
+        line = self.env['account.move.line'].search([
+            ('parent_state', '=', 'posted'),
+        ], limit=1)
+        if not line:
+            self.skipTest("No posted lines in DB")
+        period = Period(line.date, line.date, 'Single day')
+        rows = self.env['fusion.report.engine'].drill_down(
+            account_id=line.account_id.id,
+            period=period,
+            company_id=line.company_id.id,
+        )
+        self.assertIsInstance(rows, list)
+
+    def test_no_report_raises_validation_error(self):
+        period = Period(date(2026, 1, 1), date(2026, 12, 31), 'Test 2026')
+        # Inactivate any pre-existing GL definitions so the lookup
+        # fails for this test, then restore them after.
+        existing = self.env['fusion.report'].search(
+            [('report_type', '=', 'general_ledger')]
+        )
+        prior_active = {r.id: r.active for r in existing}
+        existing.write({'active': False})
+        try:
+            with self.assertRaises(ValidationError):
+                self.env['fusion.report.engine'].compute_gl(
+                    period, company_id=self.env.company.id,
+                )
+        finally:
+            for r in existing:
+                r.active = prior_active.get(r.id, True)