diff --git a/.github/requirements-security.txt b/.github/requirements-security.txt
index 11451c2..4ee81a7 100644
--- a/.github/requirements-security.txt
+++ b/.github/requirements-security.txt
@@ -1,3 +1,3 @@
 bandit==1.9.4
 pip-audit==2.10.1
-ruff==0.17.1
+ruff==0.17.3
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2829020..b8c42ba 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,8 +9,21 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ## [Unreleased]
 
+### Fixed
+
+- Native phased login accepts final sessions after any submission and skips
+  optional phases when a later state is proven. Preserve one durable attempt,
+  fresh OTP resolution and verification-only resume.
+
 ### Changed
 
+- Refresh runtime and test dependency pins, including Playwright 1.64.1.
+  Keep pyee 13.0.1 within Playwright's required >=13,<15 range so the
+  dependency batch installs successfully.
+
+- Update pypdf to 6.21.4 for seven security fixes. Align the packaged image
+  qualifier and parser documentation; preserve the `pdf-graph-v1` revision rule.
+
 - Separate durable login and document request pacing via generic engine gates. Preserve the Bourse Direct 60-second login floor; configure document pacing through Settings (default 5 seconds). No DB or recipe schema change.
 - Download All uses the recipe history limit; incremental collection keeps its
   task date window. Install the Bourse Direct recipe from the recipe catalog.
@@ -29,6 +42,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Added
 
+- Explicit `bourse_direct_adapter` catalog boundary with native login, fixed run
+  dates, C3 load/save/run-start validation, frozen trusted source and bounded
+  authenticated original reads. Existing `bourse_direct` catalogs keep their
+  current contract. The additive recipe schema must be compared and explicitly
+  promoted by Infra before a new catalog action is adopted. Provider
+  qualification and deployment remain separate.
+
 - One Infra runtime contract for consume-folder delivery, exact vault lookup,
   schema promotion, DB/configuration handoffs and daemon-path qualification.
   Provider docs and harness boundaries now refer to it. No runtime, recipe-schema
diff --git a/Dockerfile b/Dockerfile
index c83936a..b0ebe77 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -61,6 +61,10 @@ COPY --chown=${APP_UID}:${APP_GID} apps/. .
 # non-root app user
 RUN mkdir -p /apps/Downloads "$HOME" && chown -R ${APP_UID}:${APP_GID} /apps "$HOME"
 
+# Trusted catalog application code must not be group/world writable.
+RUN find /apps/recipes_playwright -type f -name 'bourse_direct_adapter.py' \
+    -exec chmod go-w {} +
+
 # Keep the schema available when recipes are mounted as mutable configuration.
 COPY apps/recipes_playwright/recipe-pw-schema.yaml /opt/billcollector/recipe-pw-schema.yaml
 
diff --git a/README.md b/README.md
index 48442bf..3ef4b90 100644
--- a/README.md
+++ b/README.md
@@ -22,7 +22,7 @@ Compose service includes a pinned seccomp profile. See
     - [Vault of Secrets](#vault-of-secrets)
     - [Enabling DNS and HTTPS with Let's Encrypt certs](#enabling-dns-and-https-with-lets-encrypt-certs)
   - [Installing the Daemon](#installing-the-daemon)
-  - [Pause notifications (ntfy)](#pause-notifications-ntfy)
+  - [Notifications (ntfy)](#notifications-ntfy)
 - [Daemon Mode](#daemon-mode)
   - [Environment Variables](#environment-variables)
   - [Container Semantics](#container-semantics)
@@ -224,7 +224,7 @@ Now that we have done a good job installing all the prerequisites, we are focusi
 
 8. Collect your first document: on the **Dashboard** press **Run now** for an enabled task - watch the run live (session page, noVNC tab) until it reaches `success`; the document appears on the Downloads page and in the consumer folder. Every step is driven from the browser, no shell access to the daemon needed.
 
-### Pause notifications (ntfy)
+### Notifications (ntfy)
 
 A fresh installation sends no notifications: the `ntfy topic` setting (`ntfy_topic` in Settings) starts empty. Paused runs, blocked logins, and failed or interrupted daemon runs only push when you set a topic. To opt in, open **Settings** and set an **ntfy topic** - choose your own topic name, it is the install-specific destination the pushes are sent to.
 
@@ -298,7 +298,7 @@ All pages require sign-in (a session cookie set on `/login`).
 - **Dashboard** (`/`) - all tasks with their status (color-coded, `waiting` rows highlighted), last/next run, latest download, a Login column (available / pending / blocked for tasks with a phased login), and a `[N waiting]` badge for paused runs; a live log pane following the active run and a step-screenshot strip; per row **Run now** (works for disabled tasks as a manual one-off, schedule untouched; disabled while the login is blocked), **Stop**, and **Edit** (cron + enabled); toolbar **Run (filtered)** (enqueues all enabled tasks in the filtered set, skipping blocked ones) and **Add task**.
 - **Task detail** (`/tasks/<id>`) - the run history of one task with the per-step records, step screenshots, and downloads; for a phased login the budget state (unproven attempts / limit) and, while idle and locked, a **Clear login block** button (grants exactly one further attempt, profile and schedule preserved).
 - **Session** (`/runs/<id>`) - the live view of one run (the ntfy deep-link target): while it runs, you watch the progress; while it is paused, you get the "Copilot" view - a screenshot, the numbered interactive elements, click/fill/press/drag actions, and Resume / Cancel - plus a noVNC tab that shows the actual browser window. A step that failed and paused also offers a Repair link into the recipe editor.
-- **Recipe editor** (`/recipes/<service>`) - edit the YAML recipe with schema validation, and optionally auto-commit the change to git (Settings switch).
+- **Recipe editor** (`/recipes/<service>`) - edit the YAML recipe with schema and method-chain validation (see [recipe method contract](doc/RECIPE_METHODS.md)), and optionally auto-commit the change to git (Settings switch).
 - **Downloads** (`/downloads`) - the downloaded files of all runs; a file whose sha256 was already seen for the same task is marked as a duplicate (the consumer hook is not re-run for it). See [download recovery](doc/DOWNLOAD_RECOVERY.md) for staged publication, recovery and the at-least-once consumer contract.
 - **Settings** (`/settings`) - UI password, ntfy topic, pause deadline, max concurrency, display start, download directory, UI base URL, VNC password, the vault API URL/host, and the git auto-commit switch. Each field writes its own Settings row on save, so an invalid value only fails that field.
 
@@ -356,7 +356,8 @@ The BillCollector configuration for each web service from which you want to retr
      - One recipe per web portal identified by the yaml element `serviceName` and its filename `recipe-pw__<serviceName>`.
      - Each recipe is structured in numbered `steps`, executed in ascending `step` order.
      - Each step is a chain of `methods`: Playwright Page/Locator calls executed in order. Locator methods (`locator` for CSS selectors, `get_by_role`, `get_by_label`, `get_by_text`, `get_by_title`, `get_by_placeholder`, `get_by_test_id`) yield a locator that is chained into the following action (`click`, `fill`, `press`, `first`). Page-level methods include `goto` (open the start URL), `content_frame` (enter an iframe) and `close`.
-     - Downloads: a step with the `expect_download` method wraps the nested `steps` that trigger the download; the file is saved into the Downloads folder. A re-downloaded file that is byte-identical to one already tracked for the same task (same sha256) is recorded as a duplicate and the consumer hook is not re-run for it.
+     - Downloads: a step with the `expect_download` method wraps the nested `steps` that trigger the download; the file is saved into the Downloads folder. A re-downloaded file that is byte-identical to one already tracked for the same task (same sha256) is recorded as a duplicate and the consumer hook is not re-run for it. An opt-in `notice_identity` step option switches a download step from byte-hash deduplication to source-scoped identity plus a conservative PDF revision digest; see [notice identity](doc/NOTICE_IDENTITY.md).
+     - Catalog adapter: use the standalone daemon-only [`bourse_direct_adapter` action](doc/BOURSE_DIRECT_ADAPTER.md) after native login. The existing locator-prefixed `bourse_direct` catalog contract remains separate.
      - Bounded pagination: use one page locator followed by `click_until_absent`. Its single options object requires `progress_selector`, `max_clicks`, and `timeout_ms`. `timeout_ms` is the total browser-action budget. Every click must increase the visible progress count or make the pager absent or hidden. This action supports cumulative "load more" controls. It rejects page-replacement pagination and virtualized lists because their visible item count does not increase. For monthly replacement calendars, use the daemon's [replacement_calendar action](doc/REPLACEMENT_CALENDARS.md).
      - Bounded collections: use one page locator followed by `download_all`. Its single options object requires `key_attribute`, `max_items`, `timeout_ms`, and `event_window_ms`. `timeout_ms` bounds the browser actions and observations for the whole collection. Persistence (`save_as`, hashing, consumers, and SQLite) remains synchronous and cannot be preempted by that browser deadline. `event_window_ms` is the full attribution window observed after each click, so `timeout_ms` must exceed the actual item count multiplied by `event_window_ms`, with additional time for browser actions. The engine reports the required and remaining budgets before its first click when the minimum attribution-window budget cannot fit. The key must be non-empty and unique for every item. The ordered key set must remain stable while each control produces exactly one download. `download_all` also rejects a suggested filename already saved earlier in the run instead of overwriting it.
      - Incremental collections: `"incremental": true` in the `download_all` options records each item key (`<key_attribute>=<key>`) with its download. Later runs of the same task skip known keys without clicking them, so only the first run collects everything. Use it only with a key that is stable per document across sessions (for example an invoice number); a constant or session-bound key misses or repeats documents. `"unpack": "pdf"` publishes the single PDF member of a ZIP download unchanged under its member name and deduplicates on the PDF; any other archive content fails without publication. Both options require the daemon runner.
@@ -386,8 +387,8 @@ The BillCollector configuration for each web service from which you want to retr
        exact-key credential substitution remain unchanged.
      - `graceful: true` on a step: a failure of that step is recorded (log + DB) and the run continues with the next step; without it, a failure aborts the service's run (after three attempts, in daemon mode the run pauses for your help instead - see [Using the Web UI](#using-the-web-ui)).
      - A `wait_user` step: an explicit pause point for operator input (e.g., a CAPTCHA you have to solve by hand). It can carry an optional `prompt` and an optional `until` locator (the pause completes when that element appears, or on resume); the step is consumed, not re-executed.
-     - There is a YAML schema named `recipe-pw-schema.yaml` which includes the rules to be followed by the YAML recipes.
-   - In daemon mode the recipe can also be edited in the UI [recipe editor](#using-the-web-ui) (schema validation, optional git auto-commit) - handy when a portal changes and a paused run links you straight to the failing recipe.
+     - There is a YAML schema named `recipe-pw-schema.yaml` which includes the rules to be followed by the YAML recipes. In daemon mode, validation goes beyond the schema: every method chain is checked against the installed Playwright (method names, argument names and types, properties, reserved `wait_user` options), and unknown methods and unsupported variables are rejected before vault or browser access, with errors that identify the service and step without argument values - the supported surface is documented in the [recipe method contract](doc/RECIPE_METHODS.md).
+   - In daemon mode the recipe can also be edited in the UI [recipe editor](#using-the-web-ui) (schema and method-chain validation, optional git auto-commit) - handy when a portal changes and a paused run links you straight to the failing recipe.
 
 > 💡 **Tip**  
 > When creating new recipes, make use of AI, e.g., let yourself be helped by Copilot - that speeds up creating the YAML recipe 🚀.
diff --git a/apps/BillCollectorServices_pw.py b/apps/BillCollectorServices_pw.py
index f1c0c53..d6d9a9a 100644
--- a/apps/BillCollectorServices_pw.py
+++ b/apps/BillCollectorServices_pw.py
@@ -285,7 +285,7 @@ def run_step(step, *, nested=False):
                             if isinstance(entry, dict)
                         }
                         if nested and methods & {
-                            "click_until_absent", "download_all", "replacement_calendar"
+                            "click_until_absent", "download_all", "replacement_calendar", "bourse_direct_adapter"
                         }:
                             raise StepAbortError(
                                 "bounded collection actions cannot be nested "
@@ -459,7 +459,7 @@ def transform_step_to_json(step):
 
     if graceful and any(
         isinstance(entry, dict)
-        and entry.get("method") in {"click_until_absent", "download_all", "replacement_calendar"}
+        and entry.get("method") in {"click_until_absent", "download_all", "replacement_calendar", "bourse_direct_adapter"}
         for entry in methods
     ):
         message = "bounded collection actions cannot be graceful"
diff --git a/apps/billcollector/bourse_direct_boundary.py b/apps/billcollector/bourse_direct_boundary.py
new file mode 100644
index 0000000..bbd1052
--- /dev/null
+++ b/apps/billcollector/bourse_direct_boundary.py
@@ -0,0 +1,288 @@
+"""One agreed catalog adapter, with native login/date/download ownership.
+
+The catalog module is trusted application code, not a sandboxed plugin. Only
+its literal CHAINS declaration is read during preflight. No import, discovery,
+registration, recipe-selected path or alternate credential engine is provided.
+"""
+
+import ast
+import copy
+import inspect
+import os
+import stat
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from urllib.parse import urlsplit
+
+from . import config
+from .collection import CollectionError
+from .interrupt import RunInterrupted
+
+ACTION = 'bourse_direct_adapter'
+IDENTITY = {'query_parameters': ['nc', 'type', 'year', 'month', 'day']}
+
+
+def _document(chains):
+    return {'services': [{'serviceName': 'BourseDirect adapter', 'steps': [
+        {'step': index, 'methods': chain}
+        for index, chain in enumerate(chains.values(), 1)]}]}
+
+
+def _validate_chains(chains):
+    from .methods import _RESERVED, validate_method_recipe
+
+    if not isinstance(chains, dict) or not chains or len(chains) > 64:
+        raise ValueError('adapter requires 1..64 named chains')
+    for name, chain in chains.items():
+        if (not isinstance(name, str) or not name.isidentifier()
+                or not isinstance(chain, list) or not 1 <= len(chain) <= 16):
+            raise ValueError('adapter chain declaration is invalid')
+        for entry in chain:
+            if (not isinstance(entry, dict) or set(entry) - {'method', 'arguments'}
+                    or not isinstance(entry.get('method'), str)
+                    or entry['method'] in _RESERVED):
+                raise ValueError('adapter chains require public Playwright methods only')
+    validate_method_recipe(_document(chains))
+
+
+@dataclass(frozen=True)
+class Candidate:
+    source: bytes
+    chains: dict
+
+
+def candidate():
+    """Freeze source without executing catalog code at load/save/run start."""
+    try:
+        path = config.RECIPE_DIR / 'bourse_direct_adapter.py'
+        metadata = path.stat()
+        if metadata.st_uid not in (0, os.getuid()) or metadata.st_mode & (stat.S_IWGRP | stat.S_IWOTH):
+            raise ValueError('adapter catalog must be root/runtime-owned and not group/world-writable')
+        source = path.read_bytes()
+        if len(source) > 256 * 1024:
+            raise ValueError
+        tree = ast.parse(source)
+        declarations = [node for node in tree.body if isinstance(node, ast.Assign)
+                        and any(isinstance(target, ast.Name) and target.id == 'CHAINS'
+                                for target in node.targets)]
+        if len(declarations) != 1:
+            raise ValueError
+        chains = ast.literal_eval(declarations[0].value)
+        functions = [node for node in tree.body if isinstance(node, ast.FunctionDef)
+                     and node.name == 'collect']
+        if (len(functions) != 1 or functions[0].decorator_list
+                or [arg.arg for arg in functions[0].args.args] != ['api', 'options']
+                or functions[0].args.defaults or functions[0].args.posonlyargs
+                or functions[0].args.kwonlyargs or functions[0].args.vararg
+                or functions[0].args.kwarg):
+            raise ValueError
+    except ValueError as error:
+        if str(error).startswith('adapter catalog must be'):
+            raise
+        raise ValueError('fixed catalog adapter source or literal CHAINS is invalid') from None
+    except (OSError, SyntaxError, TypeError, RecursionError):
+        raise ValueError('fixed catalog adapter source or literal CHAINS is invalid') from None
+    _validate_chains(chains)
+    # These are also installed Playwright calls. Check them before resources.
+    _validate_chains({'read': [
+        {'method': 'context'}, {'method': 'request'},
+        {'method': 'get', 'arguments': [{'url': 'https://example.invalid',
+                                       'max_redirects': 0, 'timeout': 1}]},
+        {'method': 'body'}]})
+    return Candidate(source, chains)
+
+
+def validate_options(options):
+    if (not isinstance(options, dict)
+            or set(options) != {'account_type', 'max_months', 'max_items', 'timeout_ms'}
+            or options['account_type'] not in ('PEA', 'PEA-PME')):
+        raise ValueError('adapter requires one supported account type and fixed bounds')
+    for name, maximum in [('max_months', 120), ('max_items', 200), ('timeout_ms', 7_200_000)]:
+        if type(options[name]) is not int or not 1 <= options[name] <= maximum:
+            raise ValueError('adapter bound is invalid')
+
+
+def prepare(recipe, *, dates=None):
+    """One standalone collection after native final-session proof, one service."""
+    found = []
+
+    def visit(step, top=False):
+        if any(entry.get('method') == 'bourse_direct' for entry in step.get('methods', [])):
+            legacy.append(step)
+        if any(entry.get('method') == ACTION for entry in step.get('methods', [])):
+            if (not top or len(step['methods']) != 1 or step.get('graceful', False)
+                    or any(key in step for key in ('steps', 'submit', 'phases'))
+                    or step.get('notice_identity') != IDENTITY):
+                raise ValueError('adapter requires one standalone notice collection step')
+            arguments = step['methods'][0].get('arguments', [])
+            if len(arguments) != 1:
+                raise ValueError('adapter requires one options object')
+            validate_options(arguments[0])
+            found.append(step)
+        for child in step.get('steps', []):
+            visit(child)
+        if 'submit' in step:
+            visit(step['submit'])
+        for phase in step.get('phases', []):
+            visit(phase)
+
+    legacy = []
+    services = recipe.get('services', [])
+    for service in services:
+        for step in service.get('steps', []):
+            visit(step, True)
+    if not found:
+        return None
+    if legacy:
+        raise ValueError('catalog adapter cannot share a task with the legacy source action')
+    if len(found) != 1 or len(services) != 1:
+        raise ValueError('adapter requires one service and one collection per task')
+    from .login import is_login
+
+    ordered = sorted(services[0]['steps'], key=lambda step: step['step'])
+    before = ordered[:ordered.index(found[0])]
+    if sum(is_login(step) for step in ordered) != 1 or not any(is_login(step) for step in before):
+        raise ValueError('adapter requires a preceding native login with final-session proof')
+    if dates is not None:
+        from .calendar import _months
+
+        try:
+            tuple(_months(dates, found[0]['methods'][0]['arguments'][0]['max_months']))
+        except CollectionError as error:
+            raise ValueError(str(error)) from None
+    return candidate()
+
+
+class Original:
+    def __init__(self, body, url, filename):
+        self.body, self.url, self.suggested_filename = body, url, filename
+
+    def save_as(self, path):
+        Path(path).write_bytes(self.body)
+
+
+def _data(value):
+    """Return data only; do not hand the adapter an unchecked browser object."""
+    if value is None or type(value) in (str, int, float, bool, bytes):
+        return value
+    if isinstance(value, (list, tuple)):
+        return [_data(item) for item in value]
+    if isinstance(value, dict) and all(isinstance(key, str) for key in value):
+        return {key: _data(item) for key, item in value.items()}
+    raise CollectionError('adapter chain must return data, not a browser object')
+
+
+class Acquisition:
+    """Capabilities for the fixed adapter. No vault, DB or login callback."""
+
+    def __init__(self, ctx, spec, options, save_download):
+        from .calendar import _months
+
+        validate_options(options)
+        self.dates = ctx.bcs.date_context
+        self.months = tuple(_months(self.dates, options['max_months']))
+        self._page, self._context = ctx.page, ctx.context
+        self._chains, self._save = copy.deepcopy(spec.chains), save_download
+        self._deadline = time.monotonic() + options['timeout_ms'] / 1000
+        self._maximum, self._offered, self._saved, self._reads = options['max_items'], 0, 0, 0
+        self._origin = self._origin_of(ctx.page.url.split('#', 1)[0])
+
+    @staticmethod
+    def _origin_of(url):
+        try:
+            parts = urlsplit(url)
+            if (parts.scheme not in ('http', 'https') or not parts.hostname
+                    or parts.username is not None or parts.password is not None
+                    or parts.fragment):
+                raise ValueError
+            return parts.scheme, parts.hostname, parts.port or (443 if parts.scheme == 'https' else 80)
+        except (ValueError, TypeError):
+            raise CollectionError('adapter source origin is invalid') from None
+
+    def remaining_ms(self):
+        remaining = int((self._deadline - time.monotonic()) * 1000)
+        if remaining < 1:
+            raise CollectionError('adapter total deadline expired')
+        return remaining
+
+    def offered(self, count):
+        """Count every offered item, including repeats, before reading a month."""
+        self.remaining_ms()
+        if type(count) is not int or count < 0 or self._offered + count > self._maximum:
+            raise CollectionError('adapter offered item limit exceeded')
+        self._offered += count
+
+    def call(self, name, **arguments):
+        """Use one declared C3 chain; dynamic kwargs replace its last arguments."""
+        if name not in self._chains:
+            raise CollectionError('adapter chain is not declared')
+        chain = copy.deepcopy(self._chains[name])
+        if arguments:
+            chain[-1]['arguments'] = [arguments]
+        _validate_chains({name: chain})
+        previous, result = self._page, None
+        for entry in chain:
+            remaining = self.remaining_ms()
+            method = getattr(previous, entry['method'])
+            args = entry.get('arguments', [])
+            positional = ['Enter'] if args == ['Enter'] else []
+            kwargs = {key: value for item in ([] if positional else args)
+                      for key, value in item.items()}
+            # A zero timeout means unbounded in Playwright. Replace it here.
+            if callable(method) and 'timeout' in inspect.signature(method).parameters:
+                kwargs['timeout'] = min(kwargs.get('timeout') or remaining, remaining)
+            result = method(*positional, **kwargs) if callable(method) else method
+            previous = result if result is not None else previous
+        self.remaining_ms()
+        return _data(result)
+
+    def read(self, url, *, pdf=False):
+        """Read unchanged bytes through the current authenticated request context."""
+        if self._origin_of(url) != self._origin or len(url) > 16384:
+            raise CollectionError('adapter request origin changed')
+        if self._reads >= self._maximum * 2:
+            raise CollectionError('adapter request limit exceeded')
+        self._reads += 1
+        response = self._context.request.get(
+            url, max_redirects=0, timeout=min(30_000, self.remaining_ms()))
+        try:
+            if response.status != 200 or response.url != url:
+                raise CollectionError('adapter response is not proven')
+            body = response.body()
+            self.remaining_ms()
+            if (not body or len(body) > 20 * 1024 * 1024
+                    or (pdf and not body.startswith(b'%PDF-'))):
+                raise CollectionError('adapter original response is invalid')
+            return body
+        finally:
+            response.dispose()
+
+    def save_original(self, url, filename):
+        if self._saved >= self._offered:
+            raise CollectionError('adapter original exceeds offered item count')
+        if (not isinstance(filename, str) or not filename.endswith('.pdf')
+                or Path(filename).name != filename or '\\' in filename):
+            raise CollectionError('adapter original filename is invalid')
+        body = self.read(url, pdf=True)
+        self._saved += 1
+        self._save(Original(body, url, filename))
+        self.remaining_ms()
+
+
+def collect(ctx, spec, options, save_download):
+    """Execute the frozen trusted catalog candidate once; redact its failures."""
+    try:
+        if spec is None:
+            raise CollectionError('adapter was not prepared before resources')
+        api = Acquisition(ctx, spec, options, save_download)
+        namespace = {'__name__': 'billcollector_bourse_direct_adapter'}
+        # Fixed operator-installed application code; never a YAML-selected path.
+        code = compile(spec.source, '<bourse_direct_adapter>', 'exec')
+        exec(code, namespace)  # nosec B102
+        namespace['collect'](api, copy.deepcopy(options))
+        api.remaining_ms()
+    except RunInterrupted:
+        raise
+    except Exception:
+        raise CollectionError('adapter acquisition failed; inspect before resuming') from None
diff --git a/apps/billcollector/collection.py b/apps/billcollector/collection.py
index 2b2333b..1b4ae74 100644
--- a/apps/billcollector/collection.py
+++ b/apps/billcollector/collection.py
@@ -24,6 +24,7 @@
 })
 COLLECTION_ACTION_METHODS = frozenset({
     "click_until_absent", "download_all", "replacement_calendar", "bourse_direct",
+    "bourse_direct_adapter",
 })
 
 logger = logging.getLogger(__name__)
@@ -53,6 +54,10 @@ class PaginationResult:
 
 def validate_terminal_action(methods, index, name):
     """Require exactly one page locator followed by one collection action."""
+    if name == 'bourse_direct_adapter':
+        if len(methods) != 1 or index != 0:
+            raise CollectionError('bourse_direct_adapter requires one standalone method')
+        return
     if len(methods) != 2 or index != 1:
         raise CollectionError(
             f"{name} requires exactly one locator method followed by the action"
diff --git a/apps/billcollector/methods.py b/apps/billcollector/methods.py
index 3116a37..d789bb5 100644
--- a/apps/billcollector/methods.py
+++ b/apps/billcollector/methods.py
@@ -16,7 +16,7 @@
 from .collection import CollectionError
 
 _RESERVED = frozenset({'login', 'wait_user', 'expect_download',
-                       'click_until_absent', 'download_all', 'replacement_calendar', 'bourse_direct'})
+                       'click_until_absent', 'download_all', 'replacement_calendar', 'bourse_direct', 'bourse_direct_adapter'})
 _UNTIL = (('role', 'get_by_role'), ('test_id', 'get_by_test_id'),
           ('placeholder', 'get_by_placeholder'), ('label', 'get_by_label'),
           ('text', 'get_by_text'), ('selector', 'locator'))
@@ -148,6 +148,39 @@ def locator_arguments(until):
 
 def validate_method_recipe(recipe):
     """Shared UI/run-start gate; diagnostics never contain argument values."""
+    from .bourse_direct_boundary import prepare
+
+    try:
+        prepare(recipe)
+    except ValueError as error:
+        def locate(service, step):
+            if any(entry.get('method') == 'bourse_direct_adapter'
+                   for entry in step.get('methods', [])):
+                return service, step
+            for child in step.get('steps', []):
+                found = locate(service, child)
+                if found:
+                    return found
+            if step.get('submit'):
+                found = locate(service, step['submit'])
+                if found:
+                    return found
+            for phase in step.get('phases', []):
+                found = locate(service, phase)
+                if found:
+                    return found
+            return None
+
+        for service in recipe.get('services', []):
+            for step in service.get('steps', []):
+                found = locate(service, step)
+                if found:
+                    owner, action = found
+                    raise ValueError(
+                        f"service {owner.get('serviceName')}, step {action.get('step')}, "
+                        f"method bourse_direct_adapter: {error}") from None
+        raise
+
     def visit(step, service):
         owners = {Page}
         for entry in step.get('methods', []):
diff --git a/apps/billcollector/notice.py b/apps/billcollector/notice.py
index d0c2180..d6bb485 100644
--- a/apps/billcollector/notice.py
+++ b/apps/billcollector/notice.py
@@ -218,7 +218,7 @@ def visit(step):
         if "notice_identity" in step:
             validate_identity(step["notice_identity"])
             names = {entry.get("method") for entry in step.get("methods", [])}
-            if not names & {"expect_download", "download_all", "replacement_calendar", "bourse_direct"}:
+            if not names & {"expect_download", "download_all", "replacement_calendar", "bourse_direct", "bourse_direct_adapter"}:
                 raise ValueError("notice_identity requires a download step")
         for nested in step.get("steps", []):
             visit(nested)
diff --git a/apps/billcollector/runner.py b/apps/billcollector/runner.py
index bfed3b9..eeb35d2 100644
--- a/apps/billcollector/runner.py
+++ b/apps/billcollector/runner.py
@@ -39,6 +39,7 @@
 import signal
 import sqlite3
 import sys
+import time
 from datetime import datetime
 
 from helpers import ServiceObj, VARIABLE_MAP
@@ -78,6 +79,7 @@
 # operator-driven resume starts a new, explicit invocation after inspection.
 _COLLECTION_METHODS = frozenset({
     "click_until_absent", "download_all", "replacement_calendar", "bourse_direct",
+    "bourse_direct_adapter",
 })
 
 
@@ -355,7 +357,7 @@ def transform_step_to_json(step):
         if method_name in [
             "click", "fill", "expect_download", "goto", "close",
             "content_frame", "first", "click_until_absent", "download_all",
-            "replacement_calendar", "bourse_direct",
+            "replacement_calendar", "bourse_direct", "bourse_direct_adapter",
         ]:
             transformed_step["actions"] = transformed_step.get("actions", [])
             transformed_step["actions"].append({
@@ -447,7 +449,7 @@ def fail_step(error_entry):
                     methods, method_index, method_name
                 )
                 processed_args = [process_argument(arg, bcs) for arg in arguments]
-                if previous_result is bcs.page:
+                if previous_result is bcs.page and method_name != 'bourse_direct_adapter':
                     raise ValueError(
                         f"{method_name} requires a locator prefix"
                     )
@@ -533,6 +535,13 @@ def bourse_direct(root, options):
                        lambda download: save_download(download, delivery_policy="bourse-direct"),
                        lambda token: process_argument(token, ctx.bcs))
 
+    def bourse_direct_adapter(_root, options):
+        from .bourse_direct_boundary import collect
+
+        if not getattr(ctx, 'native_login_proven', False):
+            raise collection.CollectionError('adapter requires native final-session proof')
+        return collect(ctx, getattr(ctx, 'bourse_direct_candidate', None), options, save_download)
+
     def save_download(download, *, source_key=None, unpack=None, delivery_policy=None):
         filename = str(download.suggested_filename)
         # An unpacked archive name is a transport name, not document identity.
@@ -594,6 +603,7 @@ def replacement_calendar(root, options):
         "download_all": download,
         "replacement_calendar": replacement_calendar,
         "bourse_direct": bourse_direct,
+        "bourse_direct_adapter": bourse_direct_adapter,
     }
 
 
@@ -830,34 +840,54 @@ def _execute_login(ctx, step):
     options = login_options(step)
     locator = _locator_for(ctx.page, process_argument(options["success"], ctx.bcs))
 
+    def record_proof():
+        if not ctx.db.login_succeeded(ctx.task["id"], ctx.run_id):
+            raise StepAbortError("Another run owns the unresolved login attempt")
+        ctx.native_login_proven = True
+        return True
+
     def proven(timeout):
         try:
             locator.wait_for(state="visible", timeout=timeout)
         except PlaywrightTimeoutError:
             return False
-        if not ctx.db.login_succeeded(ctx.task["id"], ctx.run_id):
-            raise StepAbortError("Another run owns the unresolved login attempt")
-        return True
+        return record_proof()
 
     if proven(options["session_timeout_ms"]):
         return
     phases = options["phases"]
+    markers = [_locator_for(ctx.page, process_argument(phase["success"], ctx.bcs))
+               for phase in phases[:-1]]
     start = 0
     # Find the latest completed phase before dispatching any credentials.
     for index in range(len(phases) - 2, -1, -1):
         try:
-            _locator_for(ctx.page, process_argument(phases[index]["success"], ctx.bcs)).wait_for(
-                state="visible", timeout=options["session_timeout_ms"])
+            markers[index].wait_for(state="visible", timeout=options["session_timeout_ms"])
             start = index + 1
             break
         except PlaywrightTimeoutError:
             pass
+    # After a submission, any later state may be the portal's next state.
+    # Poll all markers within one deadline, with final authentication first.
+    def next_phase(current):
+        deadline = time.monotonic() + options["timeout_ms"] / 1000
+        while True:
+            if locator.is_visible():
+                record_proof()
+                return len(phases)
+            for completed in range(len(markers) - 1, current - 1, -1):
+                if markers[completed].is_visible():
+                    return completed + 1
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise PlaywrightTimeoutError("Login has no next-state proof")
+            ctx.page.wait_for_timeout(min(50, remaining * 1000))
+
     reserved = False
     try:
-        for index in range(start, len(phases)):
+        index = start
+        while index < len(phases):
             phase = phases[index]
-            final = index == len(phases) - 1
-            marker = _locator_for(ctx.page, process_argument(phase["success"], ctx.bcs))
             for preparation in phase.get("steps", []):
                 _execute_step(ctx, preparation, parent_step=step["step"], single_attempt=True)
             if not reserved:
@@ -870,11 +900,8 @@ def proven(timeout):
                     raise StepAbortError("Login attempt refused; check the task login block")
                 reserved = True
             _execute_step(ctx, phase["submit"], parent_step=step["step"], single_attempt=True)
-            if final:
-                if proven(options["timeout_ms"]):
-                    return
-            else:
-                marker.wait_for(state="visible", timeout=options["timeout_ms"])
+            index = next_phase(index)
+        return  # next_phase proved the final authenticated session.
     except RunInterrupted:
         raise
     except Exception as e:
@@ -1096,6 +1123,13 @@ def _run_task(task_id: int, debug: bool = False, *, date_from=None, collection_m
             print(f"runner: invalid variable contract: {e}", file=sys.stderr)
             return 1
 
+        try:
+            from .bourse_direct_boundary import prepare
+            adapter_candidate = prepare(recipe, dates=dates)
+        except ValueError as e:
+            print(f"runner: invalid method contract: {e}", file=sys.stderr)
+            return 1
+
         # C6: the runner owns the per-run display: claim a display from the
         # configured range before the Run row exists (the row records it),
         # start the Xvfb/x11vnc/websockify stack, and point DISPLAY at it
@@ -1135,6 +1169,7 @@ def _run_task(task_id: int, debug: bool = False, *, date_from=None, collection_m
                 credentials = _vault_preflight(task, db)
                 ctx = _RunContext(task, run_id, db, debug, credentials, api_url)
                 ctx.bcs.date_context = dates
+                ctx.bourse_direct_candidate = adapter_candidate
                 files, aborted, pause_outcome = _run_recipe(ctx, recipe)
                 result = run_result(aborted, files, pause_outcome,
                                     known_source_count=ctx.incremental_known_count)
diff --git a/apps/billcollector/vault.py b/apps/billcollector/vault.py
index 6ea1959..ffea8cc 100644
--- a/apps/billcollector/vault.py
+++ b/apps/billcollector/vault.py
@@ -160,7 +160,7 @@ def fetch_credentials(api_url, service, user):
     """Resolve the portal credentials of one task from the vault.
 
     The item name is the name-based link ``<service> <user>`` (decision 7).
-    Returns ``{"user", "password", "otp", "url"}``; ``otp`` is None when the
+    Returns ``{"item_id", "user", "password", "otp", "url"}``; ``otp`` is None when the
     item has no TOTP entry, ``url`` is the item's first login URI
     (informational - the recipe's goto step holds the URL the engine visits).
     Raises VaultAPIError on any real failure.
diff --git a/apps/recipes_playwright/recipe-pw-schema.yaml b/apps/recipes_playwright/recipe-pw-schema.yaml
index 4c04e63..3cf77c9 100644
--- a/apps/recipes_playwright/recipe-pw-schema.yaml
+++ b/apps/recipes_playwright/recipe-pw-schema.yaml
@@ -105,6 +105,22 @@ definitions:
               additionalItems: false
           not:
             required: [ steps ]
+      - if:
+          required: [ methods ]
+          properties:
+            methods:
+              contains:
+                type: object
+                properties:
+                  method:
+                    const: bourse_direct_adapter
+                required: [ method ]
+        then:
+          properties:
+            graceful:
+              const: false
+          not:
+            required: [ steps ]
   nestedStep:
     type: object
     required: [ step ]
@@ -136,7 +152,7 @@ definitions:
                 required: [ method ]
                 properties:
                   method:
-                    enum: [ "click_until_absent", "download_all", "replacement_calendar", "bourse_direct" ]
+                    enum: [ "click_until_absent", "download_all", "replacement_calendar", "bourse_direct", "bourse_direct_adapter" ]
   method:
     oneOf:
       - $ref: "#/definitions/playwrightMethod"
@@ -144,6 +160,7 @@ definitions:
       - $ref: "#/definitions/downloadAllMethod"
       - $ref: "#/definitions/replacementCalendarMethod"
       - $ref: "#/definitions/bourseDirectMethod"
+      - $ref: "#/definitions/bourseDirectAdapterMethod"
   collectionLocatorMethod:
     type: object
     required: [ method ]
@@ -178,7 +195,7 @@ definitions:
       method:
         type: string
         not:
-          enum: [ "click_until_absent", "download_all", "replacement_calendar", "bourse_direct" ]
+          enum: [ "click_until_absent", "download_all", "replacement_calendar", "bourse_direct", "bourse_direct_adapter" ]
       arguments:
         oneOf:
           - type: array
@@ -341,6 +358,36 @@ definitions:
                 type: string
                 minLength: 1
 
+  bourseDirectAdapterMethod:
+    type: object
+    required: [method, arguments]
+    additionalProperties: false
+    properties:
+      method:
+        const: bourse_direct_adapter
+      arguments:
+        type: array
+        minItems: 1
+        maxItems: 1
+        items:
+          type: object
+          required: [account_type, max_months, max_items, timeout_ms]
+          additionalProperties: false
+          properties:
+            account_type:
+              enum: [PEA, PEA-PME]
+            max_months:
+              type: integer
+              minimum: 1
+              maximum: 120
+            max_items:
+              type: integer
+              minimum: 1
+              maximum: 200
+            timeout_ms:
+              type: integer
+              minimum: 1
+              maximum: 7200000
   bourseDirectMethod:
     type: object
     required: [method, arguments]
diff --git a/apps/requirements-dev.txt b/apps/requirements-dev.txt
index 7807cfb..bbaf996 100644
--- a/apps/requirements-dev.txt
+++ b/apps/requirements-dev.txt
@@ -1,7 +1,7 @@
 # Dev-only test dependencies (never in the runtime requirements.txt, M2 D1).
 # The suite command is: apps/.venv/bin/python -m pytest tests/ -q
-pytest==9.1.1
-pytest-asyncio==1.4.0
+pytest==9.2.0
+pytest-asyncio==1.4.1
 # Linter (config: ruff.toml at the repo root). Lint command:
 #   apps/.venv/bin/python -m ruff check apps tests
-ruff==0.17.1
+ruff==0.17.3
diff --git a/apps/requirements.txt b/apps/requirements.txt
index df8be40..d48eb67 100644
--- a/apps/requirements.txt
+++ b/apps/requirements.txt
@@ -9,12 +9,12 @@ APScheduler==3.11.3
 attrs==26.1.0
 bcrypt==5.0.0
 bidict==0.24.1
-certifi==2026.10.19
+certifi==2026.10.23
 charset-normalizer==3.5.2
 click==8.5.0
 dnspython==2.9.0
 docutils==0.23
-fastapi==0.159.1
+fastapi==0.162.0
 flatten-json==0.1.14
 frozenlist==1.8.0
 greenlet==3.5.6
@@ -30,11 +30,11 @@ jsonschema==4.26.0
 jsonschema-specifications==2025.9.1
 markdown2==2.5.5
 MarkupSafe==3.0.4
-multidict==7.1.0
+multidict==7.2.1
 nicegui==3.18.0
 opentelemetry-api==1.45.1
 orjson==3.13.0
-playwright==1.64.0
+playwright==1.64.1
 propcache==0.5.4
 pydantic==2.14.1
 pydantic_core==2.50.1
@@ -42,7 +42,7 @@ pyee==13.0.1
 pypdf==6.21.4
 Pygments==2.21.0
 python-dotenv==1.2.4
-python-engineio==4.14.0
+python-engineio==4.15.0
 python-multipart==0.0.32
 python-socketio==5.17.0
 PyYAML==6.0.3
@@ -53,11 +53,11 @@ simple-websocket==1.1.0
 six==1.17.0
 SQLAlchemy==2.1.6
 sshkeyboard==2.3.1
-starlette==1.9.0
+starlette==1.10.0
 tinycss2==1.5.1
 typing-inspection==0.4.4
 typing_extensions==4.16.0
-tzdata==2026.5
+tzdata==2026.6
 tzlocal==5.4.5
 urllib3==2.8.0
 uvicorn==0.54.0
diff --git a/doc/BOURSE_DIRECT_ADAPTER.md b/doc/BOURSE_DIRECT_ADAPTER.md
new file mode 100644
index 0000000..74fbce7
--- /dev/null
+++ b/doc/BOURSE_DIRECT_ADAPTER.md
@@ -0,0 +1,159 @@
+# Bourse Direct engine boundary
+
+This boundary implements the [issue #51 agreement](https://github.com/s-t-e-f-a-n/BillCollector/issues/51#issuecomment-6560326064).
+It does not contain or qualify a provider collector. The recipes project owns
+the portal selectors, HTML parser and source identity proofs.
+
+## Fork compatibility
+
+This fork names the new action `bourse_direct_adapter`. The public-baseline
+candidate in [PR #89](https://github.com/flowcool/BillCollector/pull/89) names
+it `bourse_direct`. This explicit difference preserves the installed catalog
+v1.0.0: its locator-prefixed `bourse_direct` action still uses the existing fork
+implementation, login, date modes and delivery policy. There is no automatic
+fallback between contracts. A task cannot combine the two source actions.
+
+The new catalog action requires native `login`. It uses the task's fixed date
+window and ordinary notice tracking. It has no invocation mode override or
+strict delivery policy. Full and incremental recipes must reuse the same task
+and history. A provider change or production switch needs separate recipes
+qualification and Infra adoption; this engine integration changes neither.
+
+## Install the candidate
+
+Install the reviewed catalog Python file as
+`apps/recipes_playwright/bourse_direct_adapter.py`, next to the recipes.
+`config.RECIPE_DIR` is the fixed installation directory. YAML cannot select a
+module, path, entry point or another provider. There is no plugin discovery or
+registration API. This is trusted application code. Review it with its recipe;
+the capabilities below are an integration interface, not a Python sandbox.
+Do not install untrusted Python through the recipe editor.
+The file must be owned by root or the runtime user and must not be group- or
+world-writable. A normal `0644` file passes this check. The image build removes
+group/world write bits from the installed adapter file.
+
+The file declares a literal `CHAINS` dictionary and `collect(api, options)`:
+
+```python
+CHAINS = {
+    'calendar_html': [
+        {'method': 'frame_locator', 'arguments': [{'selector': 'iframe#calendar'}]},
+        {'method': 'locator', 'arguments': [{'selector': 'body'}]},
+        {'method': 'inner_html'},
+    ],
+    'select_account': [
+        {'method': 'frame_locator', 'arguments': [{'selector': 'iframe#calendar'}]},
+        {'method': 'locator', 'arguments': [{'selector': 'select#nc'}]},
+        {'method': 'select_option', 'arguments': [{'value': 'synthetic'}]},
+    ],
+}
+
+def collect(api, options):
+    # Implement the provider checks described below in the recipes project.
+    raise NotImplementedError('A qualified provider adapter is required')
+```
+
+These selectors are examples, not portal evidence. The Python module may use
+standard-library parsers. Keep Playwright access in its declared chains and the
+engine's authenticated read methods. Do not import another browser, collector,
+vault client or variable engine. Do not hide collection in injected JavaScript.
+
+Load, save and run start parse the literal declaration without importing or
+executing the module. Every chain passes the installed Playwright C3 signature
+and argument gate. Reserved engine actions are prohibited inside adapter chains.
+The engine also validates its authenticated-request chain. Run start freezes the
+source and manifest before display, vault or browser access. It also rejects
+a month budget that cannot cover the fixed run-start date window before any
+resources or native login. A concurrent catalog
+change cannot replace the code of an admitted run.
+
+## Recipe
+
+Use one service and one task for one account. Before collection, navigate through
+ordinary C3 methods and execute one native `login` block. Follow
+[SAFE_LOGIN.md](SAFE_LOGIN.md) for stored sessions, password submission and optional
+fresh TOTP phases. Its final success locator must prove the authenticated session.
+The adapter receives no credential resolver, vault client, task database or login
+callback. The runner admits collection only after the native login proof succeeds.
+
+Add this standalone top-level step after that login block:
+
+```yaml
+- step: 3
+  notice_identity:
+    query_parameters: [nc, type, year, month, day]
+  methods:
+    - method: bourse_direct_adapter
+      arguments:
+        - account_type: PEA
+          max_months: 120
+          max_items: 200
+          timeout_ms: 7200000
+```
+
+The exact options are required. `account_type` is `PEA` or `PEA-PME`.
+Bounds are positive integers: at most 120 months, 200 offered items and
+7,200,000 milliseconds. No locator prefix, graceful mode, nesting or second
+collection is allowed. To migrate the historical locator -> `bourse_direct`
+recipe, rewrite it for this explicit catalog action and native login. Keeping
+the historical recipe selects the existing fork implementation. No invocation
+options are introduced for the catalog action.
+
+## Acquisition interface
+
+| Member | Contract |
+| --- | --- |
+| `api.dates` | Native immutable `DateContext`: `date_from` and `today`, fixed at run start. |
+| `api.months` | Descending `(year, month)` tuples for every complete month intersecting the inclusive date window. Rejects an insufficient month budget before adapter execution. |
+| `api.call(name, **arguments)` | Execute a declared Page-rooted chain. Optional keyword arguments replace the final method's arguments. Validate the resulting chain with C3 before dispatch. No variable expansion occurs here. Return data only, never a raw browser object. |
+| `api.remaining_ms()` | Remaining total acquisition deadline. Raises after expiry. All engine calls check it. Playwright calls with a timeout parameter receive a positive timeout capped to this deadline. |
+| `api.offered(count)` | Count all offered items before reading that month, including repeats. Reject a total above `max_items`. An empty month counts zero and still requires source proof. |
+| `api.read(url)` | Read unchanged notice HTML bytes through the current authenticated context. Require the entry page's origin, no redirects, exact response URL, HTTP 200 and 1 byte..20 MiB. Dispose each response. Total reads, including originals, are at most `2 * max_items`. |
+| `api.save_original(url, filename)` | Read the offered PDF through the same transport. Require `%PDF-` and a plain `.pdf` basename. Pass unchanged bytes and source URL to native `track_download`, with the step's notice identity. Save at most the counted offered items. |
+
+The module must not retain a browser object or implement blocking work outside
+these calls. The time budget is enforced at capability boundaries, not by a
+Python process sandbox. Calls that return API objects must end in a data read.
+Use `read` and `save_original` for source documents so their transport and bounds
+apply. Do not fetch originals through a browser chain or a separate HTTP client.
+
+Before each month, prove the offered account and month. Keep PEA and PEA-PME
+distinct. Follow offered controls. Prove an explicit empty listing. Reject
+ambiguous identity or source drift. Count every offered item. Follow notice HTML
+to its unique offered `Format PDF` URL and verify account, type and date before
+`save_original`. Read every original before notice-revision comparison. Do not
+use a daily source key to skip corrections.
+
+The step runs once automatically. A failure enters the existing runner pause;
+an operator resume is an explicit new attempt after inspection. Errors contain
+no provider values. No new pending state or receipts are added. Native Run
+results remain unchanged, including failure for a completed empty run. Native
+at-least-once delivery and `task_id:sha256` keys remain unchanged. Full and
+incremental actions use the same task, recipe and history.
+
+## Isolated engine proof
+
+Install `apps/requirements.txt` and pytest in an isolated environment. From the
+checkout root, use the browser cache that matches Playwright 1.64.0:
+
+```sh
+PLAYWRIGHT_BROWSERS_PATH=/path/to/browser-cache \
+BC_BROWSER_DIR=/path/to/browser-cache \
+<isolated-python> -m pytest tests/test_bc_bourse_direct_boundary.py \
+  tests/test_bc_method_validation.py tests/test_notice_identity.py \
+  tests/test_bc_login.py tests/test_bc_calendar.py tests/test_bc_collection.py \
+  tests/test_bc_runner.py -q
+```
+
+The boundary suite installs a synthetic catalog file in a temporary directory.
+It proves all C3 gates, rejection before resources, no preflight execution,
+frozen source, the actual adapter dispatch, native session proof, authenticated
+original reads, unchanged stored bytes, regenerated repeats, material corrections,
+source drift, empty proof, bounds and single automatic execution. Existing suites
+cover native password/TOTP, dates, collection and notice compatibility.
+No provider login, vault, production or private original is used. The recipes
+project must still qualify its actual provider file and YAML on this exact
+engine commit. The retained negative probe remains a historical-baseline proof.
+
+Recovery: revert this focused engine commit and remove this action from any
+unpublished recipe candidate. No database migration or live-state repair is needed.
diff --git a/doc/INFRA_RUNTIME_CONTRACT.md b/doc/INFRA_RUNTIME_CONTRACT.md
index f117222..1f2c4c0 100644
--- a/doc/INFRA_RUNTIME_CONTRACT.md
+++ b/doc/INFRA_RUNTIME_CONTRACT.md
@@ -96,7 +96,7 @@ Every release and Infra handoff states the SHA-256 of
 `apps/recipes_playwright/recipe-pw-schema.yaml`. This candidate's schema is:
 
 ```text
-1d3b111472c6fdfaab9860fe1ee28d6c255105c69de8281902eefcef631e70f2
+023dd141a90db23d1d6d9f182d40fe9347f7901ef0fb582f9cfa8de6a6c50e75
 ```
 
 Calculate it from the exact candidate with
diff --git a/doc/NOTICE_IDENTITY.md b/doc/NOTICE_IDENTITY.md
index 2e0998e..f78b0ae 100644
--- a/doc/NOTICE_IDENTITY.md
+++ b/doc/NOTICE_IDENTITY.md
@@ -49,7 +49,8 @@ applies. Do not use a query credential as an identity field.
 
 ## Revision rule: pdf-graph-v1
 
-The revision hashes the parsed PDF object graph with pypdf 6.21.2. It retains
+The revision hashes the parsed PDF object graph with pypdf 6.21.4. The security
+update from 6.21.2 preserves the qualified graph digests and `pdf-graph-v1`. It retains
 page content, encoded streams, fonts, images, annotations, attachments, forms,
 XMP, document metadata and local destination targets. Original bytes are never
 rewritten. The following generation differences do not create a revision:
diff --git a/doc/RECIPE_AUTHORING.md b/doc/RECIPE_AUTHORING.md
index c8af7eb..4714928 100644
--- a/doc/RECIPE_AUTHORING.md
+++ b/doc/RECIPE_AUTHORING.md
@@ -51,7 +51,7 @@ copied from another version. For example, `get_by_label` and
 `get_by_placeholder` take a `text` argument. `press` takes `key`, not `value`.
 A Page-level `click` requires `selector`; a locator's `click` does not.
 The [C3 reference](RECIPE_METHODS.md#2-dispatch-model-the-playwright-surface)
-describes the general surface and the five reserved names.
+describes the general surface and the reserved engine actions.
 
 `arguments` is a list of option objects merged into keyword arguments. Prefer
 one key per object for ordinary Playwright methods. A collection action uses
diff --git a/doc/RECIPE_METHODS.md b/doc/RECIPE_METHODS.md
index d7c90ac..242da3a 100644
--- a/doc/RECIPE_METHODS.md
+++ b/doc/RECIPE_METHODS.md
@@ -83,19 +83,22 @@ design: a locator chain such as `get_by_role` -> `first` -> `filter` ->
 `get_by_label` -> `click` is valid, and new Playwright API added by a
 version bump stays supported automatically.
 
-Six method names are **reserved**: the runner intercepts them before
+Eight method names are **reserved**: the runner intercepts them before
 Playwright dispatch and they are never passed to the browser API:
 `login`, `wait_user`, `expect_download`, `click_until_absent`,
-`download_all`, `replacement_calendar`. Their exact contracts are in section 3
-and the [replacement-calendar contract](REPLACEMENT_CALENDARS.md).
-
-A method name that is neither reserved nor a public attribute of the
-chain object's type fails at execution with `Method '<name>' not found
-on <type>`. Validation (F08) must reject such names **before** vault
-and browser access, deriving the allowlist by introspecting the pinned
-Playwright package (not a hand-maintained list), so intentionally
-supported Playwright calls are never rejected. Re-derive when the pin
-changes.
+`download_all`, `replacement_calendar`, `bourse_direct`, `bourse_direct_adapter`.
+The catalog adapter contract and legacy compatibility are in
+[BOURSE_DIRECT_ADAPTER.md](BOURSE_DIRECT_ADAPTER.md). Other contracts are in section 3
+and the [replacement-calendar contract](REPLACEMENT_CALENDARS.md). C3 validates
+calendar options with the same offline validator used by calendar execution.
+
+Validation (F08) rejects unknown attributes before vault and browser access.
+It derives the public surface, signatures and argument types from the installed
+pinned Playwright package. Re-derive the surface when the pin changes.
+Primitive and collection results can end a chain. Their Python methods cannot
+continue a recipe chain. A return annotated `Any` has no known static type;
+validation permits public Playwright continuations, while execution still checks
+the actual runtime result.
 
 Methods the shipped recipes actually use (compatibility baseline,
 section 5): `click`, `close`, `fill`, `first`, `goto`, `locator`,
@@ -242,6 +245,17 @@ publish its exact bytes, retaining any UTF-8 BOM. Unsafe paths/special entries,
 multiple files and size/count violations fail closed. Both options are rejected
 by the legacy batch adapter. See [bounds and storage](../README.md#incremental-document-collection).
 
+### Fixed catalog adapter (fork)
+
+`bourse_direct_adapter` is a standalone top-level daemon action after one native
+`login`. Its options are exactly `account_type`, `max_months`, `max_items` and
+`timeout_ms`. It requires the notice identity policy and a reviewed fixed catalog
+file with literal C3-validated `CHAINS`. It cannot be graceful, nested, preceded
+by a locator, or combined with the legacy `bourse_direct` source action.
+See [BOURSE_DIRECT_ADAPTER.md](BOURSE_DIRECT_ADAPTER.md) for the supported
+installation, limits and the naming difference from public-baseline PR #89.
+The existing `bourse_direct` recipe and delivery contracts are unchanged.
+
 ## 4. Variables
 
 The daemon expands original string argument values recursively in dictionaries
@@ -276,7 +290,7 @@ separate substitution behavior.
 ## 5. Validation requirements (F08)
 
 Gate: the daemon runner already validates the recipe before display
-allocation, vault preflight and browser launch (the shared
+allocation, vault preflight and browser launch (the shared `load_recipe` /
 `validate_recipe` gate in `runner.py::_run_task`); the UI
 validates on load and on save (`apps/billcollector/recipes.py`). The
 hardening is enforced in these same gates (`validate_recipe` / the schema /
@@ -286,8 +300,8 @@ browser access.
 Already enforced by the existing gates (keep): the schema rejects
 unknown top-level keys, missing `serviceName`/`steps`/`step`, non-
 integer step numbers, wrong argument value types, the collection
-action chain shape (exactly `[page-locator, action]`, `graceful:
-false`, no nested `steps`), the collection action argument names,
+action chain shape (exactly `[page-locator, action]`, or a standalone
+`bourse_direct_adapter`, `graceful: false`, no nested `steps`), the collection action argument names,
 types and ranges (section 3 tables), and the `phases` block shape;
 `validate_login_recipe` rejects a non-standalone, nested or
 multi-per-recipe `login`, `submit`/`phases` outside a login block, the
diff --git a/doc/SAFE_LOGIN.md b/doc/SAFE_LOGIN.md
index 5fdd41a..390ffc6 100644
--- a/doc/SAFE_LOGIN.md
+++ b/doc/SAFE_LOGIN.md
@@ -25,6 +25,11 @@ One transaction reserves one budget entry before its first submission. Each
 phase submits once. A failed phase stops all later automatic phases. Resume
 only verifies final authentication. A visible intermediate success locator lets
 the runner skip an already completed phase. Use a marker unique to that state.
+The runner checks the final session and all later phase markers after every
+submission, within one `timeout_ms` deadline. Final proof ends login immediately.
+A later intermediate marker skips optional phases that the portal did not offer.
+For example, password submission can lead directly to the final session or OTP,
+with no device chooser. The runner never fills OTP after final session proof.
 A crash between phases consumes one transaction attempt.
 An intermediate marker must disappear after the portal leaves that state. A marker
 that stays visible on later pages can make the runner skip a required phase.
@@ -142,6 +147,7 @@ The implementation follows six steps:
 
 | Requirement | Evidence in repository |
 |---|---|
+| Final session or a later phase after submission skips optional phases; no OTP after final proof | `test_optional_login_phases_follow_actual_session` |
 | Session proof bypasses login and resets budget | `test_real_browser_existing_session_skips_and_resets` |
 | One submit per phase, failed resume never replays | `test_real_browser_failure_pause_resume_never_resubmits` |
 | Password/TOTP success, failures and direct OTP entry | `test_two_phase_browser_login_counts_one_transaction` |
diff --git a/doc/vault_transport.md b/doc/vault_transport.md
index fb0881d..923ab6a 100644
--- a/doc/vault_transport.md
+++ b/doc/vault_transport.md
@@ -21,9 +21,11 @@ CLI API remains local. HTTPS is refused because `bw serve` itself speaks HTTP;
 a TLS proxy in front of it is not supported by this client.
 
 This is target enforcement, not API authentication. `bw serve` must remain on a
-trusted private network. Exact item resolution and stable account identity are
-separate work. The diagnostics fix in upstream PR #11 is also needed to prevent
-existing response/error logging from retaining sensitive data.
+trusted private network. Credential retrieval searches the item list for the
+configured name, requires exactly one exact (case-sensitive) match, and fetches
+the item and its optional TOTP by that validated item id; missing, ambiguous or
+changed identities fail with a sanitized error that retains no names, queries,
+credentials or response bodies.
 
 Verification uses `python -m unittest tests.test_vault_transport -v`, with mocked
 DNS/HTTP and a synthetic loopback partial-response server.
diff --git a/tests/qualify_bourse_direct_image.py b/tests/qualify_bourse_direct_image.py
index 62601c3..d2d8a3f 100644
--- a/tests/qualify_bourse_direct_image.py
+++ b/tests/qualify_bourse_direct_image.py
@@ -8,6 +8,7 @@
 import argparse
 import hashlib
 import importlib.util
+from importlib.metadata import distribution
 import json
 import os
 import shutil
@@ -27,11 +28,12 @@
 from billcollector import bourse_direct, config, downloads, notice, runner
 import pypdf
 assert os.getuid() == 1003 and os.getgid() == 1000
-assert pypdf.__version__ == '6.21.2'
+assert pypdf.__version__ == '6.21.4'
 shutil.copytree('/checks/tests', root/'tests')
 shutil.copytree('/apps/recipes_playwright', root/'apps/recipes_playwright')
 hashes = {name: hashlib.sha256((Path('/apps')/name).read_bytes()).hexdigest()
-          for name in ('billcollector/bourse_direct.py', 'billcollector/throttle.py', 'billcollector/notice.py',
+          for name in ('billcollector/bourse_direct.py', 'billcollector/bourse_direct_boundary.py',
+                       'billcollector/throttle.py', 'billcollector/notice.py',
                        'billcollector/downloads.py', 'billcollector/runner.py',
                        'recipes_playwright/recipe-pw-schema.yaml')}
 print(json.dumps({'packaged_source_sha256':hashes, 'uid':os.getuid(),
@@ -40,9 +42,12 @@
 os.environ['PYTEST_DISABLE_PLUGIN_AUTOLOAD'] = '1'
 sys.path.insert(0, '/test-deps')
 import pytest
-code = pytest.main([str(root/'tests/test_bourse_direct.py'),
+code = pytest.main([str(root/'tests/test_bc_login.py'),
+                    str(root/'tests/test_bourse_direct.py'),
+                    str(root/'tests/test_bc_bourse_direct_boundary.py'),
                     str(root/'tests/test_notice_identity.py'),
-                    str(root/'tests/test_throttle.py'), '-q',
+                    str(root/'tests/test_throttle.py'), '-p', 'pytest_asyncio.plugin',
+                    '-o', 'asyncio_mode=auto', '-q',
                     '--basetemp', str(root/'pytest-proof')])
 if code:
     sys.exit(code)
@@ -68,12 +73,18 @@ def main():
     deps = root/'test-deps'
     deps.mkdir(mode=0o755)
     deps.chmod(0o755)
-    for name in ('pytest', '_pytest', 'pluggy', 'iniconfig', 'packaging', 'py'):
+    for name in ('pytest', '_pytest', 'pluggy', 'iniconfig', 'packaging', 'py', 'pytest_asyncio'):
         origin = Path(importlib.util.find_spec(name).origin)
         if origin.name == '__init__.py':
             shutil.copytree(origin.parent, deps/name)
         else:
             shutil.copy2(origin, deps/origin.name)
+    # pytest-asyncio resolves its version through installed package metadata.
+    package = distribution('pytest-asyncio')
+    metadata = next(path.parent for path in package.files
+                    if path.name == 'METADATA' and path.parent.name.endswith('.dist-info'))
+    source = Path(package.locate_file(metadata))
+    shutil.copytree(source, deps / source.name)
     for path in deps.rglob('*'):
         path.chmod(0o755 if path.is_dir() else 0o644)
     root.chmod(0o755)
@@ -104,7 +115,8 @@ def main():
     state = json.loads(subprocess.run(['docker', 'inspect', container], check=True,
                                      capture_output=True, text=True).stdout)[0]
     hashes = {name: hashlib.sha256((engine/'apps'/name).read_bytes()).hexdigest()
-              for name in ('billcollector/bourse_direct.py', 'billcollector/throttle.py', 'billcollector/notice.py',
+              for name in ('billcollector/bourse_direct.py', 'billcollector/bourse_direct_boundary.py',
+                           'billcollector/throttle.py', 'billcollector/notice.py',
                            'billcollector/downloads.py', 'billcollector/runner.py',
                            'recipes_playwright/recipe-pw-schema.yaml')}
     log_text = (root/'container.log').read_text()
diff --git a/tests/test_bc_bourse_direct_boundary.py b/tests/test_bc_bourse_direct_boundary.py
new file mode 100644
index 0000000..d1206cb
--- /dev/null
+++ b/tests/test_bc_bourse_direct_boundary.py
@@ -0,0 +1,448 @@
+"""Offline proof of the engine boundary, not a qualified provider adapter."""
+
+import copy
+import io
+import json
+import threading
+from datetime import date
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from unittest.mock import Mock
+
+import pytest
+import yaml
+from playwright.sync_api import sync_playwright
+from pypdf import PdfWriter
+
+from billcollector import bourse_direct_boundary as boundary
+from billcollector import config, runner
+from billcollector.db import Database
+from billcollector.recipes import load_recipe, save_recipe, validate_recipe
+from billcollector.variables import DateContext
+
+OPTIONS = {'account_type': 'PEA', 'max_months': 3, 'max_items': 4, 'timeout_ms': 30000}
+CHAINS = {'listing': [
+    {'method': 'locator', 'arguments': [{'selector': '#listing'}]},
+    {'method': 'inner_text'}]}
+SOURCE = '''import json
+CHAINS = %r
+def collect(api, options):
+    listing = json.loads(api.call('listing'))
+    if listing['account_type'] != options['account_type']:
+        raise ValueError('private-account-drift')
+    if tuple(listing['month']) != api.months[0]:
+        raise ValueError('private-month-drift')
+    api.offered(len(listing['items']))
+    if not listing['items'] and not listing['empty']:
+        raise ValueError('private-missing-empty-proof')
+    for item in listing['items']:
+        notice = json.loads(api.read(item['notice']))
+        if len(notice['pdfs']) != 1:
+            raise ValueError('private-ambiguous-original')
+        api.save_original(notice['pdfs'][0], item['filename'])
+''' % CHAINS
+
+
+def recipe(url='http://127.0.0.1:1'):
+    return {'services': [{'serviceName': 'synthetic', 'steps': [
+        {'step': 1, 'methods': [{'method': 'goto', 'arguments': [{'url': url}]}]},
+        {'step': 2, 'methods': [{'method': 'login', 'arguments': [
+            {'success': {'selector': '#session'}, 'session_timeout_ms': 1000}]}],
+         'submit': {'step': 21, 'methods': [
+             {'method': 'locator', 'arguments': [{'selector': '#submit'}]},
+             {'method': 'click'}]}},
+        {'step': 3, 'notice_identity': copy.deepcopy(boundary.IDENTITY),
+         'methods': [{'method': boundary.ACTION, 'arguments': [copy.deepcopy(OPTIONS)]}]}]}]}
+
+
+@pytest.fixture
+def catalog(tmp_path, monkeypatch):
+    monkeypatch.setattr(config, 'RECIPE_DIR', tmp_path)
+    source = tmp_path / 'bourse_direct_adapter.py'
+    source.write_text(SOURCE)
+    source.chmod(0o644)
+    return source
+
+
+def test_load_save_and_run_start_do_not_execute_adapter(catalog, tmp_path, monkeypatch):
+    marker = tmp_path / 'should-not-exist'
+    catalog.write_text(SOURCE + f'\nopen({str(marker)!r}, "w").write("executed")\n')
+    path = tmp_path / 'recipe.yaml'
+    text = yaml.safe_dump(recipe())
+    assert validate_recipe(text)[0]
+    assert save_recipe(path, text) == (True, None)
+    assert load_recipe(path)[0]
+    monkeypatch.setattr(config, 'DB_FILE', tmp_path / 'isolated.db')
+    db = Database()
+    task = db.create_task('synthetic', 'test', recipe_file=str(path))
+    db.close()
+    monkeypatch.setattr(runner.DateContext, 'capture', Mock(return_value=DateContext(
+        date(2026, 1, 15), date(2025, 11, 15))))
+    resource = Mock(side_effect=runner.display.DisplayError('synthetic stop'))
+    monkeypatch.setattr(runner.display, 'allocate_display', resource)
+    denied = Mock(side_effect=AssertionError('must not reach vault/browser'))
+    monkeypatch.setattr(runner, '_vault_preflight', denied)
+    monkeypatch.setattr(runner, 'sync_playwright', denied)
+    assert runner.run_task(task) == 1
+    resource.assert_called_once()
+    denied.assert_not_called()
+    assert not marker.exists()
+
+
+def test_catalog_rejects_group_writable_source(catalog):
+    catalog.chmod(0o664)
+    with pytest.raises(ValueError, match='runtime-owned'):
+        boundary.candidate()
+
+
+@pytest.mark.parametrize('corrupt', ['method', 'arguments', 'literal', 'missing'])
+def test_bad_adapter_rejected_at_all_gates(catalog, tmp_path, monkeypatch, corrupt):
+    path = tmp_path / 'recipe.yaml'
+    text = yaml.safe_dump(recipe())
+    assert save_recipe(path, text)[0]
+    if corrupt == 'missing':
+        catalog.rename(catalog.with_suffix('.retained'))
+    elif corrupt == 'literal':
+        catalog.write_text(SOURCE.replace(repr(CHAINS), 'dict()'))
+    else:
+        chains = copy.deepcopy(CHAINS)
+        if corrupt == 'method':
+            chains['listing'][-1]['method'] = 'private_unknown'
+        else:
+            chains['listing'][-1]['arguments'] = [{'private_unknown': 'secret-value'}]
+        catalog.write_text(SOURCE.replace(repr(CHAINS), repr(chains)))
+    ok, error, _ = validate_recipe(text)
+    assert not ok and 'secret-value' not in error
+    assert 'service synthetic, step 3, method bourse_direct' in error
+    assert not load_recipe(path)[0]
+    assert not save_recipe(path, text + '\n# rejected')[0]
+    assert path.read_text() == text
+    monkeypatch.setattr(config, 'DB_FILE', tmp_path / 'isolated.db')
+    db = Database()
+    task = db.create_task('synthetic', 'test', recipe_file=str(path))
+    db.close()
+    denied = Mock(side_effect=AssertionError('must not acquire resources'))
+    monkeypatch.setattr(runner.display, 'allocate_display', denied)
+    monkeypatch.setattr(runner, '_vault_preflight', denied)
+    monkeypatch.setattr(runner, 'sync_playwright', denied)
+    assert runner.run_task(task) == 1
+    denied.assert_not_called()
+
+
+@pytest.mark.parametrize('change', ['prefix', 'graceful', 'nested', 'login', 'identity',
+                                  'duplicate', 'services', 'bounds', 'override'])
+def test_adapter_scope_rejected(catalog, change):
+    data = recipe()
+    steps = data['services'][0]['steps']
+    step = steps[-1]
+    if change == 'prefix':
+        step['methods'].insert(0, {'method': 'locator', 'arguments': [{'selector': 'body'}]})
+    elif change == 'graceful':
+        step['graceful'] = True
+    elif change == 'nested':
+        steps[-1] = {'step': 4, 'methods': [{'method': 'expect_download'}], 'steps': [step]}
+    elif change == 'login':
+        steps.pop(1)
+    elif change == 'identity':
+        step.pop('notice_identity')
+    elif change == 'duplicate':
+        steps.append(copy.deepcopy(step))
+    elif change == 'services':
+        data['services'].append(copy.deepcopy(data['services'][0]))
+    elif change == 'bounds':
+        step['methods'][0]['arguments'][0]['max_items'] = True
+    else:
+        step['methods'][0]['arguments'][0]['date_from'] = '2020-01-01'
+    assert not validate_recipe(yaml.safe_dump(data))[0]
+
+
+def pdf(metadata='one', width=100):
+    writer = PdfWriter()
+    writer.add_blank_page(width, 100)
+    writer.add_metadata({'/CreationDate': metadata})
+    stream = io.BytesIO()
+    writer.write(stream)
+    return stream.getvalue()
+
+
+@pytest.fixture
+def portal():
+    state = {'body': pdf(), 'requests': [], 'account_type': 'PEA', 'month': [2026, 1],
+             'empty': True, 'items': 2, 'pdfs': 1, 'status': 200}
+
+    class Handler(BaseHTTPRequestHandler):
+        def log_message(self, *_args):
+            pass
+
+        def do_GET(self):
+            state['requests'].append((self.path, self.headers.get('Cookie')))
+            base = f'http://127.0.0.1:{self.server.server_port}'
+            query = '?nc=123&type=RO&year=2026&month=1&day='
+            if self.path.startswith('/original'):
+                body = state['body']
+            elif self.path.startswith('/notice'):
+                target = base + self.path.replace('/notice', '/original') + '&pdf=1'
+                body = json.dumps({'pdfs': [target] * state['pdfs']}).encode()
+            else:
+                listing = {key: state[key] for key in ('account_type', 'month', 'empty')}
+                listing['items'] = [
+                    {'notice': base + '/notice' + query + str(day),
+                     'filename': f'original-{day}.pdf'}
+                    for day in range(1, state['items'] + 1)]
+                body = ('<div id="session">session</div><pre id="listing">'
+                        + json.dumps(listing) + '</pre>').encode()
+            self.send_response(state['status'])
+            self.send_header('Set-Cookie', 'fixture=session; Path=/')
+            self.end_headers()
+            self.wfile.write(body)
+
+    server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f'http://127.0.0.1:{server.server_port}', state
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join()
+
+
+@pytest.fixture
+def browser_ctx(catalog, portal, tmp_path, monkeypatch):
+    url, state = portal
+    monkeypatch.setattr(config, 'DB_FILE', tmp_path / 'isolated.db')
+    monkeypatch.setattr(config, 'DOWNLOAD_DIR', tmp_path / 'downloads')
+    monkeypatch.setattr(config, 'RUN_DIR', tmp_path / 'runs')
+    db = Database()
+    db.set_setting('download_dir', str(tmp_path / 'downloads'))
+    task_id = db.create_task('synthetic', 'test')
+    task = db.get_task(task_id)
+    run_id = db.create_run(task_id)
+    with sync_playwright() as p:
+        browser = p.chromium.launch(headless=True)
+        context = browser.new_context()
+        page = context.new_page()
+        page.goto(url)
+        ctx = runner._RunContext(task, run_id, db, False,
+                                 {'user': 'synthetic', 'password': 'synthetic', 'otp': None}, '')
+        ctx.page, ctx.context, ctx.bcs.page = page, context, page
+        ctx.bcs.date_context = DateContext(date(2026, 1, 15), date(2025, 11, 15))
+        ctx.bourse_direct_candidate = boundary.prepare(recipe(url))
+        monkeypatch.setattr(runner, 'pause_loop', Mock(side_effect=AssertionError('unexpected pause')))
+        # Exercise the actual native stored-session proof before collection.
+        runner._execute_login(ctx, recipe(url)['services'][0]['steps'][1])
+        monkeypatch.setattr(runner, '_record_step', Mock())
+        monkeypatch.setattr(runner.PageState, 'set_interactive_elements', lambda *_args: [])
+        try:
+            yield ctx, state
+        finally:
+            context.close()
+            browser.close()
+            db.close()
+
+
+def test_actual_adapter_path_original_bytes_and_notice_revisions(browser_ctx, catalog):
+    ctx, state = browser_ctx
+    step = recipe()['services'][0]['steps'][-1]
+    assert ctx.native_login_proven
+    # Execution uses frozen source, even if a catalog update races with the run.
+    catalog.write_text('raise AssertionError("must not execute changed source")')
+    runner._execute_step(ctx, step)
+    rows = ctx.db.conn.execute('SELECT * FROM Download').fetchall()
+    assert len(rows) == 2
+    for row in rows:
+        assert Path(row['saved_path']).read_bytes() == state['body']
+    assert all(cookie == 'fixture=session' for path, cookie in state['requests']
+               if path.startswith(('/notice', '/original')))
+    # Regenerated PDF with changed Info timestamp remains one notice revision.
+    state['body'] = pdf('two')
+    runner._execute_step(ctx, step)
+    assert ctx.db.conn.execute('SELECT count(*) FROM Download WHERE duplicate_of_id IS NULL').fetchone()[0] == 2
+    # A material correction is fetched and kept, despite the same daily source.
+    state['body'] = pdf('three', width=200)
+    runner._execute_step(ctx, step)
+    assert ctx.db.conn.execute('SELECT count(*) FROM Download WHERE duplicate_of_id IS NULL').fetchone()[0] == 4
+    assert ctx.db.conn.execute('SELECT count(*) FROM Download').fetchone()[0] == 6
+    assert len([path for path, _ in state['requests'] if path.startswith('/original')]) == 6
+    assert ctx.bcs.date_context.today == date(2026, 1, 15)
+
+
+@pytest.mark.parametrize('field,value', [('account_type', 'PEA-PME'), ('month', [2025, 12]),
+                                       ('items', 5), ('pdfs', 2), ('status', 302)])
+def test_synthetic_failure_is_single_attempt_and_redacted(browser_ctx, field, value):
+    ctx, state = browser_ctx
+    state[field] = value
+    # Update rendered listing after changing source identity/item count.
+    ctx.page.reload()
+    calls = []
+    original = boundary.collect
+
+    def count(*args):
+        calls.append(1)
+        return original(*args)
+
+    from unittest.mock import patch
+    with patch.object(boundary, 'collect', count):
+        with pytest.raises(runner.StepAbortError) as error:
+            runner._execute_step(ctx, recipe()['services'][0]['steps'][-1])
+    assert len(calls) == 1
+    assert 'private-' not in str(error.value)
+
+
+def test_explicit_empty_retains_native_empty_run_result(browser_ctx):
+    ctx, state = browser_ctx
+    state['items'] = 0
+    ctx.page.reload()
+    runner._execute_step(ctx, recipe()['services'][0]['steps'][-1])
+    assert ctx.files_downloaded == []
+    assert runner.run_result(False, ctx.files_downloaded) == 'failure'
+    state['empty'] = False
+    ctx.page.reload()
+    with pytest.raises(runner.StepAbortError):
+        runner._execute_step(ctx, recipe()['services'][0]['steps'][-1])
+
+
+def test_dates_bounds_and_runtime_call_validation(browser_ctx):
+    ctx, state = browser_ctx
+    save = Mock()
+    api = boundary.Acquisition(ctx, ctx.bourse_direct_candidate, OPTIONS, save)
+    assert api.months == ((2026, 1), (2025, 12), (2025, 11))
+    with pytest.raises(ValueError):
+        api.call('listing', unsupported='private')
+    with pytest.raises(boundary.CollectionError):
+        api.read('http://other.invalid/original')
+    with pytest.raises(boundary.CollectionError):
+        api.save_original(ctx.page.url, '../unsafe.pdf')
+    with pytest.raises(boundary.CollectionError):
+        boundary.Acquisition(ctx, ctx.bourse_direct_candidate, dict(OPTIONS, max_months=2), save)
+    api._deadline = 0
+    with pytest.raises(boundary.CollectionError):
+        api.call('listing')
+    save.assert_not_called()
+
+
+def test_full_run_start_reaches_frozen_adapter(catalog, portal, tmp_path, monkeypatch):
+    url, state = portal
+    path = tmp_path / 'recipe.yaml'
+    assert save_recipe(path, yaml.safe_dump(recipe(url)))[0]
+    monkeypatch.setattr(config, 'DB_FILE', tmp_path / 'isolated.db')
+    db = Database()
+    db.set_setting('download_dir', str(tmp_path / 'originals'))
+    task_id = db.create_task('synthetic', 'test', recipe_file=str(path))
+    db.close()
+    monkeypatch.setattr(runner.DateContext, 'capture', Mock(return_value=DateContext(
+        date(2026, 1, 15), date(2025, 11, 15))))
+    monkeypatch.setattr(runner.display, 'allocate_display', Mock(return_value=(99, 5999)))
+    monkeypatch.setattr(runner.display, 'release_display', Mock())
+    monkeypatch.setattr(runner, '_start_display_stack', Mock(return_value=[]))
+    monkeypatch.setattr(runner.display, 'stop_display_stack', Mock())
+    monkeypatch.setattr(runner, 'pause_loop', Mock(side_effect=AssertionError('unexpected pause')))
+    monkeypatch.setattr(runner, '_record_step', Mock())
+    monkeypatch.setattr(runner.PageState, 'set_interactive_elements', lambda *_args: [])
+
+    def synthetic_credentials(*_args):
+        catalog.write_text('raise AssertionError("catalog changed after admission")')
+        return {'user': 'synthetic', 'password': 'synthetic', 'otp': None}
+
+    def isolated_browser(ctx, p):
+        ctx.context = p.chromium.launch_persistent_context(str(tmp_path / 'profile'), headless=True)
+        ctx.page = ctx.context.new_page()
+        ctx.bcs.page = ctx.page
+
+    monkeypatch.setattr(runner, '_vault_preflight', synthetic_credentials)
+    monkeypatch.setattr(runner, '_launch_browser', isolated_browser)
+    assert runner.run_task(task_id) == 0
+    db = Database()
+    try:
+        assert db.conn.execute('SELECT result FROM Run').fetchone()[0] == 'success'
+        rows = db.conn.execute('SELECT saved_path FROM Download').fetchall()
+        assert len(rows) == 2
+        assert all(Path(row['saved_path']).read_bytes() == state['body'] for row in rows)
+    finally:
+        db.close()
+
+
+def test_fork_catalog_contracts_remain_explicit(catalog, monkeypatch):
+    from billcollector import bourse_direct as legacy
+
+    legacy_recipe = yaml.safe_load((Path(__file__).parent / 'fixtures/bourse_direct.yaml').read_text())
+    # Existing catalog validation never requires a new catalog module.
+    catalog.rename(catalog.with_suffix('.retained'))
+    assert validate_recipe(yaml.safe_dump(legacy_recipe))[0]
+    assert boundary.prepare(legacy_recipe) is None
+    dates = DateContext(date(2026, 1, 15), date(2025, 11, 15))
+    assert legacy.collection_dates(recipe(), dates) is dates
+    with pytest.raises(ValueError, match='Collection mode requires'):
+        legacy.select_mode(recipe(), 'all')
+    mixed = recipe()
+    mixed['services'][0]['steps'].append(legacy_recipe['services'][0]['steps'][-1])
+    with pytest.raises(ValueError, match='cannot share a task'):
+        boundary.prepare(mixed)
+
+
+def test_adapter_dispatch_never_falls_back_to_legacy(browser_ctx, monkeypatch):
+    from billcollector import bourse_direct as legacy
+
+    denied = Mock(side_effect=AssertionError('legacy provider must not run'))
+    monkeypatch.setattr(legacy, 'collect', denied)
+    ctx, _state = browser_ctx
+    runner._execute_step(ctx, recipe()['services'][0]['steps'][-1])
+    denied.assert_not_called()
+    assert len(ctx.files_downloaded) == 2
+    assert ctx.db.get_setting(f'bourse_direct_pending_{ctx.task["id"]}') in (None, '')
+
+
+def test_catalog_dispatch_requires_native_final_session(browser_ctx):
+    ctx, state = browser_ctx
+    ctx.native_login_proven = False
+    with pytest.raises(runner.StepAbortError):
+        runner._execute_step(ctx, recipe()['services'][0]['steps'][-1])
+    assert ctx.files_downloaded == []
+    assert not any(path.startswith('/original') for path, _ in state['requests'])
+
+
+@pytest.mark.parametrize('uid,accepted', [(0, True), (987654, False)])
+def test_catalog_owner_must_be_root_or_runtime(catalog, monkeypatch, uid, accepted):
+    import os
+
+    original = Path.stat
+
+    def metadata(path, **kwargs):
+        value = original(path, **kwargs)
+        if path == catalog:
+            fields = list(value)
+            fields[4] = uid
+            return os.stat_result(fields)
+        return value
+
+    monkeypatch.setattr(Path, 'stat', metadata)
+    if accepted:
+        assert boundary.candidate().source == catalog.read_bytes()
+    else:
+        with pytest.raises(ValueError, match='root/runtime-owned'):
+            boundary.candidate()
+
+
+@pytest.mark.parametrize('start', [None, '2025-01-01'])
+def test_month_budget_rejected_before_any_resources(catalog, tmp_path, monkeypatch, start):
+    path = tmp_path / 'recipe.yaml'
+    assert save_recipe(path, yaml.safe_dump(recipe()))[0]
+    monkeypatch.setattr(config, 'DB_FILE', tmp_path / 'isolated.db')
+    db = Database()
+    task_id = db.create_task('synthetic', 'test', recipe_file=str(path))
+    db.close()
+    dates = DateContext(date(2026, 1, 15), date(2025, 9, 15))
+    capture = Mock(return_value=dates)
+    monkeypatch.setattr(runner.DateContext, 'capture', capture)
+    denied = Mock(side_effect=AssertionError('must not acquire resources'))
+    monkeypatch.setattr(runner.display, 'allocate_display', denied)
+    monkeypatch.setattr(runner, '_vault_preflight', denied)
+    monkeypatch.setattr(runner, 'sync_playwright', denied)
+    assert runner.run_task(task_id, date_from=start) == 1
+    capture.assert_called_once_with(90, start)
+    denied.assert_not_called()
+    db = Database()
+    try:
+        assert not db.list_runs()
+    finally:
+        db.close()
diff --git a/tests/test_bc_login.py b/tests/test_bc_login.py
index 5c64d5c..435ee19 100644
--- a/tests/test_bc_login.py
+++ b/tests/test_bc_login.py
@@ -26,7 +26,7 @@
 from billcollector.interrupt import RunInterrupted
 from billcollector.login import login_options, settle_attempt, validate_login_recipe
 from billcollector.orchestrator import Orchestrator
-from billcollector.recipes import validate_recipe
+from billcollector.recipes import load_recipe, save_recipe, validate_recipe
 
 
 @pytest.fixture(autouse=True)
@@ -889,3 +889,67 @@ def test_finalized_attempt_recovery_ignores_reused_pid(db):
     assert Orchestrator(db).cleanup_orphans() == 0
     assert db.get_task(task)["login_pending_run_id"] is None
     assert db.get_task(task)["status"] == "idle"
+
+
+@pytest.mark.parametrize("path", ["final", "otp", "chooser", "existing"])
+@pytest.mark.parametrize("delay_ms", [0, 150])
+def test_optional_login_phases_follow_actual_session(input_login, tmp_path, path, delay_ms):
+    """A password response may finish login or skip the configured chooser."""
+    ctx, submissions, otp = input_login
+    step = input_triggered_phased_login_step()
+    chooser = {"steps": [], "success": {"selector": "#otp"},
+               "submit": {"step": 24, "methods": [
+                   {"method": "locator", "arguments": [{"selector": "#choose"}]},
+                   {"method": "click"}]}}
+    step["phases"][0]["success"] = {"selector": "#choose"}
+    step["phases"].insert(1, chooser)
+    ctx.page.evaluate("""({path, delay}) => {
+        window.choices = 0;
+        const button = document.createElement('button');
+        button.id = 'choose'; button.textContent = 'Choose'; button.hidden = true;
+        button.onclick = () => {
+            choices++; button.hidden = true;
+            document.querySelector('#otp').hidden = false;
+        };
+        document.body.append(button);
+        const old = document.querySelector('#password');
+        const password = old.cloneNode(true); old.replaceWith(password);
+        password.addEventListener('input', async () => {
+            counts[0]++;
+            await submitted('password');
+            setTimeout(() => {
+                password.hidden = true;
+                document.querySelector(path === 'final' ? '#account' :
+                    path === 'chooser' ? '#choose' : '#otp').hidden = false;
+            }, delay);
+        });
+        if (path === 'existing') document.querySelector('#account').hidden = false;
+    }""", {"path": path, "delay": delay_ms})
+    recipe = {"services": [{"serviceName": "mock", "steps": [step]}]}
+    candidate = tmp_path / "recipe-pw__optional.yaml"
+    assert save_recipe(candidate, yaml.safe_dump(recipe))[0]
+    loaded, _, _, recipe = load_recipe(candidate)
+    assert loaded
+    ctx.context = ctx.page.context
+    ctx.bcs.usr = "synthetic"
+    ctx.files_downloaded = []
+    with mock.patch.object(runner, "sync_playwright", return_value=nullcontext(None)), \
+            mock.patch.object(runner, "_launch_browser"), \
+            mock.patch.object(ctx.page, "close"), \
+            mock.patch.object(ctx.context, "close"), \
+            mock.patch.object(runner, "pause_loop", return_value="cancel") as pause:
+        _, aborted, outcome = runner._run_recipe(ctx, recipe)
+    assert not aborted and outcome is None
+    pause.assert_not_called()
+    expected_otp = int(path in {"otp", "chooser"})
+    assert ctx.page.evaluate("counts") == [int(path != "existing"), expected_otp]
+    assert ctx.page.evaluate("choices") == int(path == "chooser")
+    assert otp.call_count == expected_otp
+    assert submissions == [(phase, 1, ctx.run_id) for phase in
+                           ([] if path == "existing" else ["password"] +
+                            (["otp"] if expected_otp else []))]
+    assert ctx.native_login_proven
+    assert ctx.page.locator("#account").is_visible()
+    task = ctx.db.get_task(ctx.task["id"])
+    assert task["login_failures"] == 0
+    assert task["login_pending_run_id"] is None
diff --git a/tests/test_bc_method_validation.py b/tests/test_bc_method_validation.py
index 05fa712..02b649a 100644
--- a/tests/test_bc_method_validation.py
+++ b/tests/test_bc_method_validation.py
@@ -293,3 +293,63 @@ def test_schema_diagnostics_include_only_expected_schema_values(
     assert not ok
     assert str(expected) in error
     assert 'synthetic-secret' not in error
+
+
+@pytest.fixture
+def calendar_recipe():
+    return document([
+        {'method': 'locator', 'arguments': [{'selector': '#calendar'}]},
+        {'method': 'replacement_calendar', 'arguments': [{
+            'account_id': 'synthetic-account', 'account_selector': '.account',
+            'year_selector': '.year', 'month_selector': '.month',
+            'previous_selector': '.previous', 'item_selector': '.notice',
+            'empty_selector': '.empty', 'ready_selector': '.ready', 'max_months': 3,
+            'download_options': {'key_attribute': 'href', 'max_items': 8,
+                                 'timeout_ms': 5000, 'event_window_ms': 100},
+        }]},
+    ])
+
+
+def test_public_calendar_action_remains_valid_with_c3(calendar_recipe):
+    ok, error, _ = validate_recipe(yaml.safe_dump(calendar_recipe))
+    assert ok, error
+
+
+@pytest.mark.parametrize('mutation', ['month_names', 'selector', 'event_window'])
+def test_calendar_semantics_fail_with_secret_safe_coordinates(calendar_recipe, mutation):
+    options = calendar_recipe['services'][0]['steps'][0]['methods'][1]['arguments'][0]
+    if mutation == 'month_names':
+        options['month_names'] = ['synthetic-secret'] * 12
+    elif mutation == 'selector':
+        options['ready_selector'] = ' '
+    else:
+        options['download_options']['event_window_ms'] = 5000
+    ok, error, data = validate_recipe(yaml.safe_dump(calendar_recipe))
+    assert not ok
+    assert data is None
+    assert 'synthetic' in error and '17' in error and 'replacement_calendar' in error
+    assert 'synthetic-secret' not in error
+
+
+def test_public_notice_identity_preserves_c3_method_validation():
+    recipe = document([
+        {'method': 'locator', 'arguments': [{'selector': 'a.notice'}]},
+        {'method': 'download_all', 'arguments': [{
+            'key_attribute': 'href', 'max_items': 2, 'timeout_ms': 5000,
+            'event_window_ms': 20}]},
+    ], notice_identity={'query_parameters': ['account', 'notice']})
+    ok, error, _ = validate_recipe(yaml.safe_dump(recipe))
+    assert ok, error
+    recipe['services'][0]['steps'][0]['methods'][0]['method'] = 'unknown_method'
+    ok, error, _ = validate_recipe(yaml.safe_dump(recipe))
+    assert not ok
+    assert 'unknown_method' in error
+    assert 'synthetic' in error and '17' in error
+
+
+def test_public_notice_semantic_validation_is_kept_with_c3():
+    recipe = document([{'method': 'goto', 'arguments': [{'url': 'https://synthetic.invalid'}]}],
+                      notice_identity={'query_parameters': ['account']})
+    ok, error, _ = validate_recipe(yaml.safe_dump(recipe))
+    assert not ok
+    assert 'notice identity contract' in error
diff --git a/tests/test_bc_scheduler.py b/tests/test_bc_scheduler.py
index 0282cf9..505ee8e 100644
--- a/tests/test_bc_scheduler.py
+++ b/tests/test_bc_scheduler.py
@@ -219,5 +219,13 @@ def test_task_job_dispatches_to_the_run_callback(self):
         self.assertEqual(self.calls, [task_id])
 
 
+def test_ui_and_scheduler_share_numeric_weekday_validation():
+    from billcollector.ui.tasks import _valid_cron
+    for day in ('0', '7', 'sun', '1-5', '*/2'):
+        assert _valid_cron(f'17 4 * * {day}')
+    for day in ('8', '1/2', 'mon-5', '01', '*/02'):
+        assert not _valid_cron(f'17 4 * * {day}')
+
+
 if __name__ == "__main__":
     unittest.main()