diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 922c50e..73e197e 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -75,6 +75,6 @@ jobs: - uses: Swatinem/rust-cache@v2 - run: pip install -e ".[dev,postgres]" - name: PostgreSQL journal, recovery and mocked venue contracts - run: pytest -q tests/test_postgres_storage.py tests/test_owner_migrations.py + run: pytest -q tests/test_postgres_storage.py tests/test_owner_migrations.py tests/test_routed_execution_scope.py tests/test_routed_scope_transactions.py tests/test_routed_audit_pagination.py env: SYNPATH_TEST_POSTGRES_DSN: postgresql://postgres:postgres@localhost:5432/synpath_test diff --git a/synpath/engine/README.md b/synpath/engine/README.md index c4076df..0fa7542 100644 --- a/synpath/engine/README.md +++ b/synpath/engine/README.md @@ -55,12 +55,14 @@ separate execution/authentication design. ## Schema and migration compatibility -The current logical version is 10. Alembic revision `0001_journal_v6` adopts the +The current logical version is 11. Alembic revision `0001_journal_v6` adopts the frozen pre-Alembic baseline; `0002_single_user` adds the owner and foreign keys; `0003_bucket_member_enabled` adds `bucket_members.enabled` (existing members stay on); `0004_orders_recent` indexes `orders` newest first, overall and per market, and gives any row without a `created_ts` its `updated_ts`; `0005_drop_fair_values` drops -the unused `fair_values` table. +the unused `fair_values` table. `0006_routed_execution_scope` adds indexed +execution-account membership for routed parents and removes their nominal venue +from snapshots and readable projections. See the upgrade preflight below. See the [migration entry point](migrations/README.md). Historical SQLite conversions prepare old files for the baseline; new changes use new revisions. @@ -179,3 +181,63 @@ all venue guarantees. International Polymarket eligibility remains blocked on th previous test host. Polymarket/US do not echo client keys; automatic adoption after response loss needs persisted native identities. US ordinary orders lack general reduce-only support. Its already-eligible test is not evidence of a future crossing. + + +## Cross-venue parent identity and upgrade preflight + +A bucket order is owned by the engine's stable `user_id`, with `venue=null` and +`account=null`. `execution_accounts` captures each enabled venue's exact Account; +venue children keep their concrete venue/account. A single captured bucket/account +scope is authorized, risk-checked and saved before routing; later bucket edits do +not alter it. One transaction commits the snapshot, account membership, readable +projection, submission result and events. The private `__managed__` storage key +must never be configured as a venue. Normal venue orders retain their concrete +identity. Without an explicit account mapping, the engine uses an adapter's +reported Account when available, otherwise the venue's default Account. + +Bucket requests cannot name a singular account or venue. Configure per-venue +accounts instead. Admission requires full-quantity capacity at every eligible +member (including correctly flipped outcome/price), then every actual child +passes current risk checks again. A valid split can therefore be refused when +no individual destination has capacity for the whole quantity. This reserves +neither funds nor capacity. Existing parent and venue states are unchanged. + +Viewing a complete parent/report or its audit history requires visibility of all +captured accounts; acting on it requires trade permission on all of them. Child +records retain their individual account access. `/journal`, `/order-log`, live +events and replay use the same parent-scope rules. Caller-provided request params +never establish audit visibility. Old order-list cursors must be discarded; the +new opaque cursor has version 2. Parent lookup works after the order finishes +and the process restarts; venue/account filters use captured membership. + +Back up and verify restore before upgrading. Migration preserves user, parent, +client and child IDs, status and fills, and records old nominal attribution as +`legacy_execution_identity`. It infers historical accounts only from the saved +parent account and persisted child orders/intents. When an enabled venue has no +unambiguous historical account evidence, startup refuses the migration and names +the affected parent/venue. Do not replace it with today's configured account. + +For a reviewed historical mapping, while the old engine is stopped, store a JSON +object under `meta.key='legacy_execution_accounts'`: parent ID -> venue -> complete +Account object (`venue`, `name`, optional `subaccount`). For example, an explicitly +reviewed mapping has shape `{"mo-example":{"polymarket":{"venue":"polymarket", +"name":"historical-desk","subaccount":null}}}`. This is an operator decision, +not an automatically generated guess. Insert/update that meta row in the existing +backend with bound values, retain the review evidence, then start the new engine +through its normal locked migration path. Conflicting historical evidence still +fails; an explicit mapping cannot silently override a persisted account. Missing +and conflicting cases roll back and can be retried after review on both backends. + +A legacy client ID without a verifiable submission reservation is refused rather +than submitted again; retrieve its existing parent by ID. New bucket client-ID +retries use the original captured scope even after a bucket edit. A mismatched +configured/adapter account blocks affected submissions, early uncertain-intent +recovery, child refresh, polling, reconciliation and cancellation. The original +activity stays unresolved until the original execution identity is available; +absence in a replacement account cannot establish that an order is lost. + +Do not run older code on the migrated store: restore a compatible backup, with +outstanding venue activity reviewed first. Do not run both engines concurrently. +This upgrade does not resolve the broader uncertain-send/resync limitations in +[the routed lifecycle](orders/lifecycle/routed.md). All PRE-6 verification uses +offline scripted venues; this documentation does not authorize actual trades. diff --git a/synpath/engine/engine.py b/synpath/engine/engine.py index 9fa440c..b0d518e 100644 --- a/synpath/engine/engine.py +++ b/synpath/engine/engine.py @@ -42,7 +42,7 @@ import uuid from dataclasses import dataclass, field from decimal import Decimal -from typing import Any, Iterable, Mapping +from typing import Any, Callable, Iterable, Mapping from .. import ids from ..bucket import Bucket, bucket_id_of, is_bucket_id @@ -54,6 +54,7 @@ Settlement, Side, Precision, ) from ..trading.instruments import MarketMaster +from .execution_scope import MANAGED_NAMESPACE, BucketNotFound, validate_execution_accounts from .events import EventBus from .journal import IntentState, LeaseLost, now_ms from .storage import JournalStore, create_journal @@ -192,6 +193,8 @@ def __init__( matcher: Matcher | None = None, ): self.config = config or EngineConfig() + if MANAGED_NAMESPACE in adapters or MANAGED_NAMESPACE in (accounts or {}): + raise ValueError("the managed storage namespace cannot be configured as a venue") self.adapters: dict[str, TradingExchange] = dict(adapters) if not self.config.user_id.strip(): raise ValueError("user_id must not be empty") @@ -207,7 +210,9 @@ def __init__( for venue in self.adapters: # Every configured venue trades under some account; name the # default one, so listing accounts shows what orders will use. - self.accounts.setdefault(venue, Account(venue=venue)) + adapter_account = getattr(self.adapters[venue], "account", None) + self.accounts.setdefault(venue, adapter_account if isinstance(adapter_account, Account) + else Account(venue=venue)) self.clock = clock self.running = False self.started_ts: int | None = None @@ -297,7 +302,8 @@ async def __aexit__(self, *exc: Any) -> None: # -- submitting ----------------------------------------------------------- async def submit(self, request: OrderRequest, *, venue: str | None = None, risk_reference_price: Decimal | None = None, - precision: Precision | None = None) -> Order: + precision: Precision | None = None, + authorize_execution_accounts: Callable[[dict[str, Account]], None] | None = None) -> Order: """Retry the same instruction, never repeat an uncertain venue send. `risk_reference_price` overrides the book midpoint for the limit-price risk check. @@ -309,8 +315,13 @@ async def submit(self, request: OrderRequest, *, venue: str | None = None, risk_ request = request.model_copy(update={"user_id": self.config.user_id}) if request.type in (OrderType.RFQ_TAKER, OrderType.RFQ_MAKER): raise BadRequest(f"a {request.type.value} order comes from a quote: use accept_quote or confirm_quote") + if is_bucket_id(request.market_id): + if venue is not None: + raise BadRequest("a bucket parent has no singular execution venue") + return await self._submit_bucket(request, risk_reference_price=risk_reference_price, + authorize_execution_accounts=authorize_execution_accounts) supported = VENUE_ORDER_TYPES | {OrderType.STOP_MARKET, OrderType.STOP_LIMIT} - if is_bucket_id(request.market_id) or request.type not in supported or request.params.get("walk"): + if request.type not in supported or request.params.get("walk"): return await self._submit_once(request, venue=venue, risk_reference_price=risk_reference_price, precision=precision) venue = venue or self._venue_of(request) account = request.account or self.accounts.get(venue) or Account(venue=venue) @@ -469,38 +480,122 @@ async def _submit_once(self, request: OrderRequest, *, venue: str | None = None, await self.bus.publish("order.accepted", order.model_dump(mode="json"), key=f"{venue}:{order.id}", persist=False) return order - async def _submit_bucket(self, request: OrderRequest, *, risk_reference_price: Decimal | None) -> Order: - """An order on a bucket is a market order the engine holds as a - `routed_limit` parent. The request stores the enabled-member bucket - configuration captured at submission; later edits to the saved bucket - do not change this running order. Its `price` is the worst price the - caller accepts, in bucket terms, as on any market order here. The - nominal venue is the first member's; the legs each carry their own.""" + async def _submit_bucket(self, request: OrderRequest, *, risk_reference_price: Decimal | None, + authorize_execution_accounts: Callable[[dict[str, Account]], None] | None = None) -> Order: + """Capture, authorize and persist one cross-venue instruction. + + The mutable bucket is read once. A reservation retains the captured + bucket/account scope if acceptance is interrupted before parent creation. + Retrying an accepted client ID returns its original parent. + """ if request.type != OrderType.MARKET: - raise BadRequest("an order on a bucket is a market order: set type to market, " - "and price to the worst price you accept") + raise BadRequest("an order on a bucket is a market order: set type to market") if request.price is None: - raise BadRequest("a market order on a bucket needs price: the worst price you accept, in bucket terms") - row = await self.journal.bucket(bucket_id_of(request.market_id)) - if row is None: - raise BadRequest(f"{request.market_id}: no such bucket in this journal") - bucket = Bucket.model_validate(row) - if bucket.status != "active": - raise BadRequest(f"{request.market_id}: the bucket is {bucket.status}") - # Capture the bucket configuration for this order, keeping only enabled members. - # The parent uses this configuration throughout execution, including after restart; - # later edits to the saved bucket do not change which members this order trades. - # Disabled members receive no child orders and require no venue adapters here. - # The saved bucket definition remains unchanged. - bucket = bucket.routable() - missing = bucket.venues() - set(self.adapters) - if missing: - raise BadRequest(f"{request.market_id}: no adapter configured for {', '.join(sorted(missing))}") - nominal = ids.venue_of(bucket.members[0].market_id) - account = request.account or self.accounts.get(nominal) or Account(venue=nominal) - params = {**request.params, "managed_kind": "routed_limit", "bucket": bucket.model_dump(mode="json")} - request = request.model_copy(update={"params": params, "account": account}) - return await self.submit_managed(request, venue=nominal, account=account, risk_reference_price=risk_reference_price) + raise BadRequest("a market order on a bucket needs price: the worst price you accept") + if request.account is not None: + raise BadRequest("a bucket parent has no singular account; configure each venue's execution account") + request = request.model_copy(update={"client_order_id": request.client_order_id or f"sp-{uuid.uuid4().hex[:20]}"}) + async with self._lock(f"submission:{request.client_order_id}"): + previous = await self.journal.submission(request.client_order_id) + if previous is not None: + if previous["request"] != request_identity(request): + raise BadRequest("client_order_id already identifies a different instruction") + effective = OrderRequest.stored(previous["effective_request"]) + accounts = validate_execution_accounts({venue: Account.model_validate(account) + for venue, account in effective.params["execution_accounts"].items()}) + if authorize_execution_accounts: + authorize_execution_accounts(accounts) + result = await self._existing_submission_result(previous) + if result is not None: + return result + else: + if await self.journal.order_by_client_id(request.client_order_id) is not None: + raise BadRequest("client_order_id belongs to an older bucket order without a verifiable instruction; look it up by parent ID") + row = await self.journal.bucket(bucket_id_of(request.market_id)) + if row is None: + raise BucketNotFound(f"{request.market_id}: no such bucket in this journal") + bucket = Bucket.model_validate(row) + if bucket.status != "active": + raise BadRequest(f"{request.market_id}: the bucket is {bucket.status}") + bucket = bucket.routable() + missing = bucket.venues() - set(self.adapters) + if missing: + raise BadRequest(f"no adapter configured for {', '.join(sorted(missing))}") + accounts = validate_execution_accounts({venue: self.accounts.get(venue) or Account(venue=venue) + for venue in bucket.venues()}) + if authorize_execution_accounts: + authorize_execution_accounts(accounts) + params = {**request.params, "managed_kind": "routed_limit", "bucket": bucket.model_dump(mode="json"), + "execution_accounts": {venue: account.model_dump(mode="json") for venue, account in accounts.items()}} + effective = request.model_copy(update={"params": params}) + await self.journal.record_submission(request, effective, venue=MANAGED_NAMESPACE, account=None) + try: + bucket = Bucket.model_validate(effective.params["bucket"]) + self._evaluate_bucket_risk(effective, bucket, accounts) + return await self.submit_managed(effective, venue=None, account=None, + execution_accounts=accounts, risk_reference_price=risk_reference_price) + except (RiskRejected, BadRequest) as error: + await self.journal.reject_submission(request.client_order_id, + {"kind": type(error).__name__, "message": str(error), "rule": getattr(error, "rule", None)}) + raise + + def _evaluate_bucket_risk(self, request: OrderRequest, bucket: Bucket, accounts: dict[str, Account]) -> None: + """Check the instruction once, then full capacity in each possible destination. + + Destinations are alternatives, not summed exposure. Only actual venue + children count as resting venue orders; parent projections are not orders + at an exchange. No capacity is reserved by these admission checks. + """ + decision = self.order_risk_checks.evaluate_order_risk(request, venue=None, account=None, instruction_only=True) + if not decision.ok: + raise RiskRejected(decision.message, rule=decision.rule) + resting = [order for order in self.open_orders() if order.held_by == HeldBy.VENUE] + for member in sorted(bucket.members, key=lambda item: item.market_id): + venue = ids.venue_of(member.market_id) + self.assert_execution_account(venue, accounts[venue]) + outcome = request.outcome + if member.flip and outcome is not None: + outcome = "no" if outcome == "yes" else "yes" + projected = request.model_copy(update={"market_id": member.market_id, + "side": (Side.SELL if request.side == Side.BUY else Side.BUY) if member.flip else request.side, + "price": Decimal("1") - request.price if member.flip else request.price, + "outcome": outcome, "account": accounts[venue], "tags": {**request.tags, "parent": "admission"}}) + decision = self.order_risk_checks.evaluate_order_risk(projected, venue=venue, account=accounts[venue], + open_orders=[order for order in resting if order.venue == venue], + market_close_ts=self.market_close.get(member.market_id), event_id=self.event_of.get(member.market_id), + venue_notional=sum((abs(state.contracts) for (key, _, _), state in self.ledger.positions.items() + if key == accounts[venue].key), ZERO)) + if not decision.ok: + raise RiskRejected(f"{venue}/{accounts[venue].key}: {decision.message}", rule=decision.rule) + + def assert_execution_account(self, venue: str, account: Account) -> None: + """Refuse account-bound I/O when a saved routed account no longer matches. + + The engine supports one configured execution identity per venue. It + cannot reconcile or cancel the old account through replacement credentials. + """ + configured = self.accounts.get(venue) or Account(venue=venue) + adapter_account = getattr(self.adapters.get(venue), "account", None) + if (account.venue != venue or account != configured + or (isinstance(adapter_account, Account) and adapter_account != account)): + raise SubmissionPending(f"{venue}: saved execution account {account.key} differs from configured " + f"{configured.key}; restore the original account before recovery or execution") + + async def guard_routed_order_account(self, order: Order) -> None: + parent_id = order.parent_id or order.tags.get("parent") + if parent_id: + stored = await self.journal.managed_one(parent_id) + if stored and stored[0] == "routed_limit": + if order.account is None: + raise SubmissionPending("routed child has no persisted execution account") + self.assert_execution_account(order.venue, order.account) + + async def guard_routed_venue_accounts(self, venue: str) -> None: + for saved_key in await self.journal.routed_account_keys(venue): + configured = self.accounts.get(venue) or Account(venue=venue) + if saved_key != configured.key: + raise SubmissionPending(f"{venue}: saved routed account {saved_key} differs from configured {configured.key}") + self.assert_execution_account(venue, configured) async def save_bucket(self, bucket: Bucket) -> Bucket: """Store or update a bucket definition in this engine's journal.""" @@ -508,13 +603,14 @@ async def save_bucket(self, bucket: Bucket) -> Bucket: await self.journal.save_bucket(bucket.model_dump(mode="json")) return bucket - async def submit_managed(self, request: OrderRequest, *, venue: str, account: Account, + async def submit_managed(self, request: OrderRequest, *, venue: str | None, account: Account | None, + execution_accounts: dict[str, Account] | None = None, risk_reference_price: Decimal | None = None) -> Order: """Accept an engine-held order: check it, journal it, start it.""" check_managed_fields(request, now_ms=int(self.clock() * 1000)) - decision = self.evaluate_order_risk( + decision = (Decision(ok=True) if execution_accounts else self.evaluate_order_risk( request, venue=venue, account=account, risk_reference_price=risk_reference_price, - ) + )) if not decision.ok: await self.bus.publish("risk.rejected", { "rule": decision.rule, "message": decision.message, "market_id": request.market_id, @@ -523,7 +619,7 @@ async def submit_managed(self, request: OrderRequest, *, venue: str, account: Ac raise RiskRejected(decision.message, rule=decision.rule) request = request.model_copy(update={"user_id": request.user_id or self.config.user_id, "client_order_id": request.client_order_id or f"sp-{uuid.uuid4().hex[:20]}"}) - parent = await self.managed_order_controller.create(request, venue=venue, account=account) + parent = await self.managed_order_controller.create(request, venue=venue, account=account, execution_accounts=execution_accounts) return parent.as_order() async def submit_child(self, parent: Any, request: OrderRequest, **kw: Any) -> Order: @@ -535,7 +631,13 @@ async def submit_child(self, parent: Any, request: OrderRequest, **kw: Any) -> O no venue falls back to the parent's, which is every single-venue type.""" venue = self._child_venue(parent, request) account = parent.account - if venue != parent.venue: + if parent.execution_accounts: + account = parent.execution_accounts.get(venue) + if account is None: + raise BadRequest(f"{venue}: child is outside the saved execution scope") + self.assert_execution_account(venue, account) + request = request.model_copy(update={"account": account}) + elif venue != parent.venue: # `child_request` stamps the parent's account on every child; a # leg on another venue must carry that venue's instead. account = self.accounts.get(venue) or Account(venue=venue) @@ -552,6 +654,8 @@ def _child_venue(parent: Any, request: OrderRequest) -> str: try: return ids.venue_of(request.market_id) except BadRequest: + if parent.venue is None: + raise BadRequest("a routed child must name its concrete venue") from None return parent.venue async def cancel_child(self, parent: Any, order_id: str, venue: str | None = None) -> Order | None: @@ -579,7 +683,7 @@ def evaluate_order_risk( risk_reference_price = self.book_mid(market_id) return self.order_risk_checks.evaluate_order_risk( request, venue=venue, account=account, - open_orders=[o for o in self.open_orders(venue=venue)], + open_orders=[o for o in self.open_orders(venue=venue) if not o.execution_accounts], risk_reference_price=risk_reference_price, market_close_ts=self.market_close.get(market_id), event_id=self.event_of.get(market_id), caller_supplied_order_id=caller_supplied_order_id, @@ -602,6 +706,7 @@ async def cancel(self, order_id: str, *, venue: str | None = None, market_id: st # engine-held parent this process no longer runs because it # finished before a restart (live parents are always restored). return order + await self.guard_routed_order_account(order) venue = venue or order.venue adapter = self._adapter(venue) key = f"cancel:{venue}:{order_id}" @@ -610,6 +715,7 @@ async def cancel(self, order_id: str, *, venue: str | None = None, market_id: st price=order.price, book=order.book, trader=order.trader, client_order_id=key, ) account = order.account or self.accounts.get(venue) or Account(venue=venue) + request = request.model_copy(update={"account": account, "tags": dict(order.tags)}) await self.journal.record_intent(request, client_order_id=key, venue=venue, account=account, operation="cancel", target_order_id=order_id) async with self._lock(key): @@ -659,6 +765,9 @@ async def edit(self, request: EditRequest, *, venue: str | None = None) -> Order """Amend a resting order. The edit carries its own idempotency key, so a retry is the same edit rather than a second one.""" order = await self._known(request.order_id, venue) + if order.held_by == HeldBy.ENGINE: + raise BadRequest("engine-held parents cannot be amended; cancel and submit a new instruction") + await self.guard_routed_order_account(order) venue = venue or order.venue adapter = self._adapter(venue) key = request.client_order_id or f"edit:{venue}:{request.order_id}:{uuid.uuid4().hex[:8]}" @@ -675,6 +784,7 @@ async def edit(self, request: EditRequest, *, venue: str | None = None) -> Order "rule": decision.rule, "message": decision.message, "order_id": order.id, "operation": "edit", }, key=key) raise RiskRejected(decision.message, rule=decision.rule) + intent_request = intent_request.model_copy(update={"account": account, "tags": dict(order.tags)}) await self.journal.record_intent(intent_request, client_order_id=key, venue=venue, account=account, operation="edit", target_order_id=order.id) async with self._lock(key): @@ -762,6 +872,7 @@ async def halt(self, reason: str, *, scope: str = "*", policy: HaltPolicy | None if scope not in ("*", venue): continue try: + await self.guard_routed_venue_accounts(venue) count = await adapter.cancel_all_orders() result["canceled"][venue] = count except Exception as exc: # a venue that cannot be reached must not stop the others @@ -771,7 +882,8 @@ async def halt(self, reason: str, *, scope: str = "*", policy: HaltPolicy | None if left: result["remaining"][venue] = left for order in self.open_orders(): - if scope in ("*", order.venue): + if (order.held_by == HeldBy.VENUE and scope in ("*", order.venue) + and not isinstance(result["canceled"].get(order.venue), str)): canceled = order.model_copy(update={"status": OrderStatus.CANCELED}) await self.journal.upsert_order(canceled, event="order.canceled") self._remember(canceled) @@ -870,6 +982,17 @@ async def _resolve(self, intent: Any) -> dict[str, Any]: await self.bus.publish("intent.unresolved", base | {"reason": "no adapter for this venue"}) return base | {"result": "unresolved", "reason": "no adapter"} try: + persisted = OrderRequest.stored(intent.request) + parent_id = persisted.tags.get("parent") + stored = await self.journal.managed_one(parent_id) if parent_id else None + if stored and stored[0] == "routed_limit": + if persisted.account is None: + raise SubmissionPending("routed intent has no saved execution account") + self.assert_execution_account(intent.venue, persisted.account) + if intent.target_order_id: + target = await self.journal.order(intent.venue, intent.target_order_id) + if target is not None: + await self.guard_routed_order_account(target) found = await self._find_by_client_id(adapter, intent) except Exception as exc: await self.bus.publish("intent.unresolved", base | {"reason": f"{type(exc).__name__}: {exc}"}) @@ -927,7 +1050,7 @@ async def _find_by_client_id(self, adapter: TradingExchange, intent: Any) -> Ord def open_orders(self, *, venue: str | None = None, book: str | None = None) -> list[Order]: """Open orders as the engine believes them, from memory of the journal.""" return [o for o in self._open_cache.values() - if (venue is None or o.venue == venue) and (book is None or o.book == book)] + if (venue is None or o.venue == venue or venue in o.execution_accounts) and (book is None or o.book == book)] async def refresh_open_orders(self) -> list[Order]: orders = await self.journal.open_orders() @@ -1197,6 +1320,7 @@ async def _poll_loop(self) -> None: log.exception("synpath.engine: polling %s failed", venue) async def poll(self, venue: str, adapter: TradingExchange | None = None) -> None: + await self.guard_routed_venue_accounts(venue) adapter = adapter or self._adapter(venue) for order in await adapter.fetch_open_orders(): await self.on_order(order) @@ -1232,6 +1356,11 @@ def _venue_of(self, request: OrderRequest) -> str: "use a Synpath id (venue:native), pass venue=, or name an account") async def _known(self, order_id: str, venue: str | None) -> Order: + managed = await self.journal.managed_one(order_id) + if managed is not None: + parent = self._rebuilt(*managed) + if parent is not None: + return parent.as_order() if venue is not None: order = await self.journal.order(venue, order_id) if order is not None: diff --git a/synpath/engine/execution_scope.py b/synpath/engine/execution_scope.py new file mode 100644 index 0000000..5891124 --- /dev/null +++ b/synpath/engine/execution_scope.py @@ -0,0 +1,34 @@ +"""Execution accounts and private storage identity for routed instructions. + +The managed namespace is a database key, never a venue or an Account. +Only children have a concrete execution venue; a routed parent saves the +complete set of accounts it is allowed to use. +""" +from __future__ import annotations + +from ..trading.types import Account, Order +from ..errors import BadRequest + +MANAGED_NAMESPACE = "__managed__" + + +class BucketNotFound(BadRequest): + """A missing bucket remains a bad SDK request and an HTTP 404.""" + + +def storage_identity(order: Order) -> tuple[str, str]: + if order.execution_accounts: + return MANAGED_NAMESPACE, MANAGED_NAMESPACE + return order.venue, order.account.key if order.account else f"{order.venue}:default" + + +def order_accounts(order: Order) -> list[str]: + if order.execution_accounts: + return sorted(account.key for account in order.execution_accounts.values()) + return [order.account.key if order.account else f"{order.venue}:default"] + + +def validate_execution_accounts(accounts: dict[str, Account]) -> dict[str, Account]: + if not accounts or any(venue != account.venue for venue, account in accounts.items()): + raise ValueError("execution accounts must match their concrete venues") + return dict(sorted(accounts.items())) diff --git a/synpath/engine/journal.py b/synpath/engine/journal.py index 3156b5f..b4564b9 100644 --- a/synpath/engine/journal.py +++ b/synpath/engine/journal.py @@ -41,7 +41,9 @@ from .submissions import request_identity from .order_storage import OrderRecords, UNATTRIBUTED_USER, schema as order_schema -SCHEMA_VERSION = 10 +from .execution_scope import MANAGED_NAMESPACE, storage_identity + +SCHEMA_VERSION = 11 SCHEMA = """ CREATE TABLE IF NOT EXISTS submissions ( @@ -590,24 +592,25 @@ async def submission(self, client_order_id: str) -> dict[str, Any] | None: return result async def record_submission(self, original: OrderRequest, effective: OrderRequest, - *, venue: str, account: Account) -> None: + *, venue: str, account: Account | None) -> None: """Reserve the instruction and its audit event in one transaction.""" stamp = now_ms() identity = request_identity(original) + account_key = account.key if account else MANAGED_NAMESPACE async with self._write: await self._db.execute("BEGIN IMMEDIATE") try: cursor = await self._db.execute( "INSERT INTO submissions(client_order_id,user_id,account_key,venue,request,effective_request,created_ts,updated_ts) " "VALUES(?,?,?,?,?,?,?,?) ON CONFLICT(client_order_id) DO NOTHING", - (original.client_order_id, original.user_id, account.key, venue, + (original.client_order_id, original.user_id, account_key, venue, _dump(identity), _dump(effective.model_dump(mode="json")), stamp, stamp), ) if cursor.rowcount: await self._db.execute( "INSERT INTO events(ts,kind,key,payload) VALUES(?,?,?,?)", (stamp, "submission.reserved", original.client_order_id, - _dump({"user_id": original.user_id, "account": account.key, "venue": venue, + _dump({"user_id": original.user_id, "account": account_key, "venue": venue, "request": identity, "effective_request": effective.model_dump(mode="json")})), ) else: @@ -751,46 +754,49 @@ def _intent_of(row: Any) -> Intent: # -- orders and fills ----------------------------------------------------- async def upsert_order(self, order: Order, *, event: str = "order.updated") -> None: - stamp = now_ms() - body = order.model_dump(mode="json") - account_key = order.account.key if order.account else f"{order.venue}:default" async with self._write: await self._db.execute("BEGIN IMMEDIATE") try: - await self._db.execute( - """INSERT INTO orders(venue, id, client_order_id, account_key, market_id, side, - status, terminal, book, trader, quantity, filled, price, parent_id, payload, - created_ts, updated_ts) - VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) - ON CONFLICT(venue, id) DO UPDATE SET - client_order_id=COALESCE(excluded.client_order_id, orders.client_order_id), - status=excluded.status, terminal=excluded.terminal, filled=excluded.filled, - price=excluded.price, quantity=excluded.quantity, payload=excluded.payload, - book=COALESCE(excluded.book, orders.book), trader=COALESCE(excluded.trader, orders.trader), - parent_id=COALESCE(excluded.parent_id, orders.parent_id), - updated_ts=excluded.updated_ts""", - (order.venue, order.id, order.client_order_id, account_key, order.market_id, - _word(order.side), _word(order.status), 1 if order.is_terminal else 0, order.book, order.trader, - str(order.quantity), str(order.filled), _text(order.price), - order.parent_id or order.tags.get("parent"), _dump(body), - order.created_at or stamp, stamp), - ) - if order.held_by.value == "venue": - await self._order_records.write_order(order, stamp=stamp) - await self._db.execute( - "UPDATE submissions SET result_id=?, state='accepted', updated_ts=? " - "WHERE client_order_id=? AND user_id=? AND account_key=?", - (order.id, stamp, order.client_order_id, order.user_id, account_key), - ) - await self._db.execute( - "INSERT INTO events(ts, kind, key, payload) VALUES(?,?,?,?)", - (stamp, event, f"{order.venue}:{order.id}", _dump(body)), - ) + await self._write_order(order, event=event, stamp=now_ms()) await self._db.execute("COMMIT") except BaseException: await self._db.execute("ROLLBACK") raise + async def _write_order(self, order: Order, *, event: str, stamp: int) -> None: + """Write the order projection inside the caller's transaction.""" + body = order.model_dump(mode="json") + storage_venue, account_key = storage_identity(order) + await self._db.execute( + """INSERT INTO orders(venue, id, client_order_id, account_key, market_id, side, + status, terminal, book, trader, quantity, filled, price, parent_id, payload, + created_ts, updated_ts) + VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(venue, id) DO UPDATE SET + client_order_id=COALESCE(excluded.client_order_id, orders.client_order_id), + status=excluded.status, terminal=excluded.terminal, filled=excluded.filled, + price=excluded.price, quantity=excluded.quantity, payload=excluded.payload, + book=COALESCE(excluded.book, orders.book), trader=COALESCE(excluded.trader, orders.trader), + parent_id=COALESCE(excluded.parent_id, orders.parent_id), + updated_ts=excluded.updated_ts""", + (storage_venue, order.id, order.client_order_id, account_key, order.market_id, + _word(order.side), _word(order.status), 1 if order.is_terminal else 0, order.book, order.trader, + str(order.quantity), str(order.filled), _text(order.price), + order.parent_id or order.tags.get("parent"), _dump(body), + order.created_at or stamp, stamp), + ) + if order.held_by.value == "venue": + await self._order_records.write_order(order, stamp=stamp) + await self._db.execute( + "UPDATE submissions SET result_id=?, state='accepted', updated_ts=? " + "WHERE client_order_id=? AND user_id=? AND account_key=?", + (order.id, stamp, order.client_order_id, order.user_id, account_key), + ) + await self._db.execute( + "INSERT INTO events(ts, kind, key, payload) VALUES(?,?,?,?)", + (stamp, event, f"{storage_venue}:{order.id}", _dump(body)), + ) + async def record_fill(self, fill: Fill, *, book: str | None = None, trader: str | None = None) -> bool: """Store a fill. `False` means this fill id was already known, which happens on every venue that delivers at least once.""" @@ -828,7 +834,8 @@ async def record_fill(self, fill: Fill, *, book: str | None = None, trader: str raise return changed - async def order(self, venue: str, order_id: str) -> Order | None: + async def order(self, venue: str | None, order_id: str) -> Order | None: + venue = venue or MANAGED_NAMESPACE async with self._db.execute("SELECT payload FROM orders WHERE venue=? AND id=?", (venue, order_id)) as cursor: row = await cursor.fetchone() return Order.model_validate(json.loads(row["payload"])) if row else None @@ -845,8 +852,13 @@ async def open_orders(self, *, venue: str | None = None, account_key: str | None args: list[Any] = [] for column, value in (("venue", venue), ("account_key", account_key), ("book", book)): if value is not None: - sql += f" AND {column}=?" - args.append(value) + if column in ("venue", "account_key"): + sql += (f" AND ({column}=? OR (orders.venue=? AND EXISTS " + f"(SELECT 1 FROM parent_execution_accounts scope WHERE scope.parent_id=orders.id AND scope.{column}=?)))") + args.extend([value, MANAGED_NAMESPACE, value]) + else: + sql += f" AND {column}=?" + args.append(value) async with self._db.execute(sql + " ORDER BY updated_ts", args) as cursor: return [Order.model_validate(json.loads(row["payload"])) async for row in cursor] @@ -863,8 +875,13 @@ async def orders_page(self, *, venue: str | None = None, book: str | None = None args: list[Any] = [] for column, value in (("venue", venue), ("book", book), ("market_id", market_id)): if value is not None: - sql += f" AND {column}=?" - args.append(value) + if column in ("venue", "account_key"): + sql += (f" AND ({column}=? OR (orders.venue=? AND EXISTS " + f"(SELECT 1 FROM parent_execution_accounts scope WHERE scope.parent_id=orders.id AND scope.{column}=?)))") + args.extend([value, MANAGED_NAMESPACE, value]) + else: + sql += f" AND {column}=?" + args.append(value) if since_ts is not None: sql += " AND created_ts >= ?" args.append(since_ts) @@ -988,25 +1005,30 @@ async def archive_bucket(self, bucket_id: str) -> bool: # -- engine-held orders --------------------------------------------------- - async def save_managed(self, snapshot: dict[str, Any]) -> None: + async def save_managed(self, snapshot: dict[str, Any], *, order: Order | None = None) -> None: """Commit the authoritative snapshot, shared projection and event.""" async with self._write: await self._db.execute("BEGIN IMMEDIATE") try: - await self._write_managed(snapshot, stamp=now_ms()) + stamp = now_ms() + await self._write_managed(snapshot, stamp=stamp) + if order is not None: + await self._write_order(order, event=f"managed.{snapshot['state']}", stamp=stamp) await self._db.execute("COMMIT") except BaseException: await self._db.execute("ROLLBACK") raise async def plan_managed_child(self, snapshot: dict[str, Any], request: OrderRequest, - *, venue: str, account: Account) -> None: + *, venue: str, account: Account, parent_order: Order | None = None) -> None: """The irreversible parent transition and child dispatch commit together.""" async with self._write: await self._db.execute("BEGIN IMMEDIATE") try: stamp = now_ms() await self._write_managed(snapshot, stamp=stamp) + if parent_order is not None: + await self._write_order(parent_order, event=f"managed.{snapshot['state']}", stamp=stamp) await self._write_intent(request, client_order_id=request.client_order_id, venue=venue, account=account, stamp=stamp) await self._db.execute("COMMIT") @@ -1036,6 +1058,20 @@ async def _write_managed(self, snapshot: dict[str, Any], *, stamp: int) -> None: _dump(snapshot), children[-1]["order_id"] if children else None, int(snapshot.get("created_ms") or stamp), stamp), ) + if snapshot.get("execution_accounts"): + async with self._db.execute("SELECT venue,account_key FROM parent_execution_accounts WHERE parent_id=?", + (snapshot["id"],)) as cursor: + existing_scope = {row["venue"]: row["account_key"] async for row in cursor} + proposed_scope = {venue: Account.model_validate(body).key + for venue, body in snapshot["execution_accounts"].items()} + if existing_scope and existing_scope != proposed_scope: + raise ValueError("a routed parent's saved execution scope is immutable") + for venue, body in snapshot["execution_accounts"].items(): + execution_account = Account.model_validate(body) + await self._db.execute( + "INSERT INTO parent_execution_accounts(parent_id,venue,account_key) VALUES(?,?,?) " + "ON CONFLICT(parent_id,venue) DO NOTHING", + (snapshot["id"], venue, execution_account.key)) await self._order_records.write_managed(snapshot, user_id=self.user_id, stamp=stamp) await self._db.execute( "INSERT INTO events(ts, kind, key, payload) VALUES(?,?,?,?)", @@ -1066,6 +1102,14 @@ async def managed_on(self, market_id: str) -> list[tuple[str, dict[str, Any]]]: ) as cursor: return [(row["kind"], json.loads(row["payload"])) async for row in cursor] + async def routed_account_keys(self, venue: str) -> list[str]: + """Indexed live-parent account scope for polling/reconciliation guards.""" + async with self._db.execute( + "SELECT DISTINCT scope.account_key FROM parent_execution_accounts scope " + "JOIN managed_orders parent ON parent.id=scope.parent_id " + "WHERE scope.venue=? AND parent.state IN ('waiting','working','cancelling')", (venue,)) as cursor: + return [row["account_key"] async for row in cursor] + async def managed_one(self, parent_id: str) -> tuple[str, dict[str, Any]] | None: async with self._db.execute(f"SELECT kind, payload FROM ({MANAGED_SOURCE}) AS managed_source WHERE id=?", (parent_id,)) as cursor: row = await cursor.fetchone() diff --git a/synpath/engine/migrations/README.md b/synpath/engine/migrations/README.md index ad1b662..19ee990 100644 --- a/synpath/engine/migrations/README.md +++ b/synpath/engine/migrations/README.md @@ -1,7 +1,7 @@ # Journal migrations Alembic revisions run through the engine's locked SQLAlchemy connection. Current -head is `0005_drop_fair_values`, logical schema version 10. Configuration, ownership +head is `0006_routed_execution_scope`, logical schema version 11. Configuration, ownership compatibility and backup/restore behavior live in the [engine README](../README.md). - [0001 journal baseline](versions/0001_journal_v6.py): frozen version-6 schema. @@ -9,6 +9,7 @@ compatibility and backup/restore behavior live in the [engine README](../README. - [0003 bucket member enabled](versions/0003_bucket_member_enabled.py): `bucket_members.enabled`, existing members on. - [0004 orders recent](versions/0004_orders_recent.py): `orders_recent` and `orders_market_recent` indexes for listing every order newest first; undated rows take `updated_ts`. - [0005 drop fair values](versions/0005_drop_fair_values.py): the unused `fair_values` table; positions are marked from the live book. +- [0006 routed execution scope](versions/0006_routed_execution_scope.py): captured parent account membership and venue-neutral routed snapshots/projections; missing historical accounts require a reviewed mapping before upgrade. Add a reviewed revision for schema/data changes; never edit deployed revisions or import mutable runtime definitions into them. Verify both databases and rollback diff --git a/synpath/engine/migrations/versions/0006_routed_execution_scope.py b/synpath/engine/migrations/versions/0006_routed_execution_scope.py new file mode 100644 index 0000000..e5d5177 --- /dev/null +++ b/synpath/engine/migrations/versions/0006_routed_execution_scope.py @@ -0,0 +1,98 @@ +"""Replace nominal parent attribution with captured execution membership. + +Migration definitions are frozen. Missing historical accounts require an +explicit mapping in meta['legacy_execution_accounts'] keyed by parent ID, +then venue. Never substitute the current process's configured accounts. +""" +import json + +from alembic import op +import sqlalchemy as sa + +revision = "0006_routed_execution_scope" +down_revision = "0005_drop_fair_values" +branch_labels = None +depends_on = None +NAMESPACE = "__managed__" + + +def account_key(account): + return f"{account['venue']}:{account.get('name', 'default')}" + ( + f":{account['subaccount']}" if account.get('subaccount') else "") + + +def upgrade(): + connection = op.get_bind() + op.create_table("parent_execution_accounts", + sa.Column("parent_id", sa.Text, sa.ForeignKey("managed_orders.id"), primary_key=True), + sa.Column("venue", sa.Text, primary_key=True), + sa.Column("account_key", sa.Text, nullable=False)) + op.create_index("parent_execution_venue", "parent_execution_accounts", ["venue", "parent_id"]) + op.create_index("parent_execution_account", "parent_execution_accounts", ["account_key", "parent_id"]) + metadata = sa.MetaData() + parents = sa.Table("managed_orders", metadata, autoload_with=connection) + orders = sa.Table("orders", metadata, autoload_with=connection) + intents = sa.Table("intents", metadata, autoload_with=connection) + membership = sa.Table("parent_execution_accounts", metadata, autoload_with=connection) + meta = sa.Table("meta", metadata, autoload_with=connection) + supplied = connection.execute(sa.select(meta.c.value).where(meta.c.key == "legacy_execution_accounts")).scalar() + mappings = json.loads(supplied) if supplied else {} + historical_intent_accounts = {} + for payload in connection.execute(sa.select(intents.c.request)).scalars(): + child_request = json.loads(payload) + parent_id = (child_request.get("tags") or {}).get("parent") + if parent_id: + historical_intent_accounts.setdefault(parent_id, []).append(child_request.get("account")) + for row in connection.execute(sa.select(parents).where(parents.c.kind == "routed_limit")).mappings().all(): + snapshot = json.loads(row["payload"]) + request = snapshot["request"] + venues = {member["market_id"].split(":", 1)[0] + for member in request["params"]["bucket"]["members"] if member.get("enabled", True)} + accounts = {} + + def remember(account): + if not account or account.get("venue") not in venues: + return + venue = account["venue"] + if venue in accounts and account_key(accounts[venue]) != account_key(account): + raise ValueError(f"{row['id']}: conflicting historical execution accounts; explicit review required") + accounts[venue] = account + + remember(snapshot.get("account")) + children = connection.execute(sa.select(orders.c.payload).where(orders.c.parent_id == row["id"])).scalars() + for payload in children: + remember(json.loads(payload).get("account")) + # Uncertain children may have an intent but no returned order yet. + for account in historical_intent_accounts.get(row["id"], []): + remember(account) + for venue, account in mappings.get(row["id"], {}).items(): + if account.get("venue") != venue or venue not in venues: + raise ValueError(f"{row['id']}: invalid explicit execution account mapping") + remember(account) + if set(accounts) != venues: + missing = ", ".join(sorted(venues - set(accounts))) + raise ValueError(f"{row['id']}: historical execution accounts missing for {missing}; " + "review and populate meta.legacy_execution_accounts before upgrade") + accounts = dict(sorted(accounts.items())) + snapshot["legacy_execution_identity"] = {"venue": snapshot.get("venue"), "account": snapshot.get("account")} + snapshot.update(venue=None, account=None, execution_accounts=accounts) + request["account"] = None + request["params"]["execution_accounts"] = accounts + connection.execute(parents.update().where(parents.c.id == row["id"]).values( + account_key=None, payload=json.dumps(snapshot))) + for venue, account in accounts.items(): + connection.execute(membership.insert().values(parent_id=row["id"], venue=venue, account_key=account_key(account))) + projections = connection.execute(sa.select(orders).where( + orders.c.id == row["id"], orders.c.parent_id.is_(None))).mappings().all() + for projection in projections: + payload = json.loads(projection["payload"]) + if payload.get("held_by") != "engine": + continue + payload.update(venue=None, account=None, execution_accounts=accounts) + connection.execute(orders.update().where(orders.c.venue == projection["venue"], orders.c.id == row["id"]).values( + venue=NAMESPACE, account_key=NAMESPACE, payload=json.dumps(payload))) + connection.execute(meta.update().where(meta.c.key == "schema_version").values(value="11")) + + +def downgrade(): + raise RuntimeError("execution-scope downgrade requires an explicit compatible backup/restore plan") diff --git a/synpath/engine/order_log.py b/synpath/engine/order_log.py index 2278946..c25a0cd 100644 --- a/synpath/engine/order_log.py +++ b/synpath/engine/order_log.py @@ -147,7 +147,10 @@ async def order_keys(engine: Any, order_id: str) -> list[str]: if managed is not None: _, snapshot = managed venue = snapshot.get("venue") - keys += [f"{venue}:{current}"] if venue else [] + keys += [f"{venue}:{current}"] if venue else [f"__managed__:{current}"] + legacy_venue = (snapshot.get("legacy_execution_identity") or {}).get("venue") + if legacy_venue: + keys.append(f"{legacy_venue}:{current}") client = snapshot.get("client_order_id") or (snapshot.get("request") or {}).get("client_order_id") keys += [client] if client else [] family += [str(child.get("order_id") if isinstance(child, dict) else child) @@ -160,7 +163,7 @@ async def order_keys(engine: Any, order_id: str) -> list[str]: async def order_log(engine: Any, *, order_id: str | None = None, limit: int = 200, - before: int | None = None) -> tuple[list[dict[str, Any]], int | None]: + before: int | None = None, allow_event=None) -> tuple[list[dict[str, Any]], int | None]: """The log newest first, for one order (and its family) or the whole account, with the `before` cursor for the next page (`None` at the end). Raw events are read in batches until the folded page is full: a waiting @@ -171,7 +174,8 @@ async def order_log(engine: Any, *, order_id: str | None = None, limit: int = 20 while True: batch = await engine.journal.events_page(kinds=ORDER_EVENT_KINDS, prefixes=ORDER_EVENT_PREFIXES, keys=keys, before=cursor, limit=RAW_BATCH) - events += [event for event in batch if event.kind not in NOT_ORDER_KINDS] + events += [event for event in batch if event.kind not in NOT_ORDER_KINDS + and (allow_event is None or await allow_event(event))] exhausted = len(batch) < RAW_BATCH if batch: cursor = batch[-1].seq diff --git a/synpath/engine/orders/base.py b/synpath/engine/orders/base.py index cdec96e..0aae70f 100644 --- a/synpath/engine/orders/base.py +++ b/synpath/engine/orders/base.py @@ -188,7 +188,8 @@ def last_trade_timestamp(self, market_id: str) -> int | None: async def plan_child(self, request: OrderRequest) -> None: await self.engine.journal.plan_managed_child(self.parent.snapshot(), request, - venue=self.parent.venue, account=self.parent.account) + venue=self.parent.venue, account=self.parent.account, + parent_order=self.parent.as_order()) async def observe_pending_child(self, client_order_id: str) -> tuple[Order | None, Intent | None]: order = await self.engine.journal.order_by_client_id(client_order_id) @@ -228,14 +229,17 @@ def __init__( request: OrderRequest, *, id: str | None = None, - venue: str, - account: Account, + venue: str | None, + account: Account | None, + execution_accounts: dict[str, Account] | None = None, state: State = "waiting", ): self.id = id or f"mo-{uuid.uuid4().hex[:16]}" self.request = request self.venue = venue self.account = account + self.execution_accounts = execution_accounts or {} + self.legacy_execution_identity = None self.state: State = state self.children: list[Child] = [] self.created_ms = now_ms() @@ -252,8 +256,8 @@ def market_id(self) -> str: return self.request.market_id @property - def account_key(self) -> str: - return self.account.key + def account_key(self) -> str | None: + return self.account.key if self.account else None @property def side(self) -> Side: @@ -295,7 +299,7 @@ def venues(self) -> set[str]: """Every venue this parent has or may put a child on. A single-venue parent is its own venue; a parent across venues overrides this so a halt scoped to any one of them reaches it.""" - return {self.venue} | {c.venue for c in self.children} + return ({self.venue} if self.venue else set()) | set(self.execution_accounts) | {c.venue for c in self.children} # -- hooks ---------------------------------------------------------------- @@ -480,10 +484,13 @@ def snapshot(self) -> dict[str, Any]: """Everything needed to rebuild this parent after a restart.""" return { "id": self.id, "kind": self.kind, "state": self.state, "venue": self.venue, - "account": self.account.model_dump(mode="json"), "request": self.request.model_dump(mode="json"), + "account": self.account.model_dump(mode="json") if self.account else None, + "execution_accounts": {venue: account.model_dump(mode="json") for venue, account in self.execution_accounts.items()}, + "request": self.request.model_dump(mode="json"), "filled": str(self.filled), "children": [c.to_dict() for c in self.children], "created_ms": self.created_ms, "updated_ms": self.updated_ms, "detail": self.detail, "owner_id": self.owner_id, "extra": self.extra(), + "legacy_execution_identity": self.legacy_execution_identity, } def extra(self) -> dict[str, Any]: @@ -497,12 +504,16 @@ def load_extra(self, extra: dict[str, Any]) -> None: def from_snapshot(cls, snapshot: Mapping[str, Any]) -> "ManagedOrder": request = OrderRequest.stored(snapshot["request"]) parent = cls(request, id=snapshot["id"], venue=snapshot["venue"], - account=Account.model_validate(snapshot["account"]), state=snapshot["state"]) + account=Account.model_validate(snapshot["account"]) if snapshot.get("account") else None, + execution_accounts={venue: Account.model_validate(account) + for venue, account in (snapshot.get("execution_accounts") or {}).items()}, + state=snapshot["state"]) parent.children = [Child.of(row) for row in snapshot.get("children") or []] parent.created_ms = int(snapshot.get("created_ms") or now_ms()) parent.updated_ms = int(snapshot.get("updated_ms") or parent.created_ms) parent.detail = snapshot.get("detail") or "" parent.owner_id = snapshot.get("owner_id") + parent.legacy_execution_identity = snapshot.get("legacy_execution_identity") parent.load_extra(snapshot.get("extra") or {}) return parent @@ -516,6 +527,7 @@ def as_order(self) -> Order: average = sum((child.filled * child.average_price for child in filled_children), ZERO) / self.filled return Order( user_id=self.request.user_id, id=self.id, client_order_id=self.request.client_order_id, venue=self.venue, account=self.account, + execution_accounts=dict(self.execution_accounts), market_id=self.market_id, side=self.side, type=self.order_type, time_in_force=self.request.time_in_force, status=STATUS_OF[self.state], held_by=HeldBy.ENGINE, price=self.request.price, stop_price=self.request.stop_price, quantity=self.quantity, filled=self.filled, diff --git a/synpath/engine/orders/lifecycle/nominal-venue-removal-design.md b/synpath/engine/orders/lifecycle/nominal-venue-removal-design.md new file mode 100644 index 0000000..36f58e6 --- /dev/null +++ b/synpath/engine/orders/lifecycle/nominal-venue-removal-design.md @@ -0,0 +1,281 @@ +# PRE-6: cross-venue parent identity and execution scope + +Status: approved for implementation on 25 October 2026, including D6/D7 independent-review additions. Offline verification and pre-landing review are required before merge. No actual trades are authorized. + +Issue: https://linear.app/synpath/issue/PRE-6/remove-arbitrary-nominal-venue-from-cross-venue-smart-order-parents + +## Baselines and destinations + +- prediction_market worktree: `/Users/slingshot/.codex/worktrees/pre-6-cross-venue-parent/prediction_market`, based on main `cda4369a804a96f54288966ab96cf2438ef2de10`; eventual PR base `main`. +- Public SDK worktree: `/private/tmp/pre-6-synpath-public`, branch `codex/pre-6-cross-venue-parent`, based on experimental branch `85013870850c5741788a99a0c746fe224de2e366`; eventual PR base `codex/order-lifecycle-storage`. +- The original checkouts contain unrelated work and remain untouched. +- The current terminal is a read-only market-data application with execution previews. It has no order submission endpoint. Execution UI on another unmerged branch is not assumed to be part of main. +- The latest experimental SDK branch is the terminal compatibility policy. Record tested commits in evidence; do not introduce a workspace SHA pin or update production package versions. + +## Shareable explanation: old uses and replacements + +1. **Account selection.** The parent took the first enabled member's venue and its account, and child requests initially inherited that account; children on other venues then replaced it. The parent will instead belong to the configured user, with a saved map of eligible venues to execution accounts. Each child selects its account directly from that map using its own market ID, independent of member order. +2. **Risk checks.** The parent currently runs a single-venue risk check against the nominal venue/account, although its children can execute elsewhere. The replacement separates checks on the whole instruction from checks on all eligible execution scopes, followed by the existing checks on each actual child. Flipped members use their own side and complemented YES price for member-specific checks. +3. **Permissions.** Bucket submission already checks member-venue permissions, but parent reads, amendments and cancellations use its one nominal account. The replacement checks the saved account scope consistently: viewing the whole parent requires visibility of every account; actions affecting the whole parent require trade permission on every account. Later bucket edits cannot change those permissions for an existing parent. +4. **Storage and lookup.** The nominal venue currently supplies the parent projection's database key and participates in account filters, pagination and cached lookup. The replacement uses the stable parent ID and a private managed-order storage namespace, with an explicit saved execution scope for account and venue membership queries. Existing parent rows and snapshots require a reviewed migration; venue children keep their existing keys and attribution. +5. **Reporting and lifecycle.** Public parent responses currently show the nominal venue/account even when all fills occur elsewhere. The replacement reports no single execution venue/account, lists eligible venues/accounts explicitly, and continues to derive actual execution and fills from children. Cancellation targets each child's real venue, while a halt of any eligible venue reaches the whole parent, as the routed parent's existing `venues()` method already intends. + +## Proposed contract and ownership + +Use the existing user identity; `owner_id` on ManagedOrder already means a nested managed-order relationship and must not be repurposed as user ownership. Keep book and trader attribution unchanged. + +A routed parent, including a bucket with only one enabled member, has `venue=null`, `account=null`, and an explicit execution scope containing a deterministically ordered venue-to-Account map. This scope is captured with the enabled bucket definition before routing begins and saved in the durable snapshot. Existing single-venue parents and venue orders retain their concrete venue/account. Validate that a venue-held order always has a concrete venue; nullable parent fields must never reach adapters, fills, or ledger keys. + +Reject a caller-supplied singular account for a new bucket instruction with an actionable error: it cannot specify the accounts of several venues. Initially use the engine's configured per-venue accounts, retaining the established default-account convention where applicable. A new API for selecting several accounts is outside this ticket. Validate each selected account's venue against its member venue and the configured execution adapter; do not claim arbitrary subaccount routing when an adapter only supports one configured identity. + +Persist the exact account selection; a restart with conflicting execution configuration must block new dispatch for that parent with an actionable diagnostic rather than silently select a different account. Preserve access to historical records and the ability to reconcile or clean up through the original accounts when those adapters are available. + +D4 accepted on 25 October 2026: prepare the enabled bucket and its account map once in the trusted submission flow. HTTP permission checks, parent risk, durable persistence and routing consume that same prepared snapshot; the engine must not reread the mutable bucket after authorization. Preparation is internal, never trusted from a caller-supplied account-scope payload. Direct SDK calls reuse the preparation path while retaining the existing execution-owner checks. Bind any supported same-client-ID retry to the saved original instruction/scope, not a newly edited bucket. Test a bucket edit interleaved between authorization and dispatch. + +```text +user + bucket instruction + client order ID + | + enabled bucket snapshot + saved venue -> account map + | + instruction / scope validation + authorization + parent risk + | + durable routed parent + | + existing fee-aware allocator + / \ + member market A flipped member B + venue/account A venue/account B + child risk/send converted child risk/send + \ / + child status / fills + | + aggregate parent and report +``` + +## Risk policy agreed in architecture review + +D2 accepted on 25 October 2026: conservative admission requires capacity for the full requested quantity at every eligible destination. Keep the existing risk rechecks on actual children; admission does not reserve capacity or guarantee execution. + +Run instruction-wide rules once: user/book validity, price and quantity bounds, order notional, applicable duplicate/rate controls and daily loss. Check the bucket restriction itself and every enabled member restriction. Check venue halts across all enabled venues before accepting the instruction, even when today's best price would allocate only to one venue. + +For member/account risk, project the full remaining quantity to each eligible member as a conservative admission check: any eligible member may receive that quantity after a price change. Convert flipped side, outcome and price into member coordinates, and use that member's account, market-close time, event, positions and open orders. Alternative destinations are alternatives, not simultaneous allocations: do not add a full parent quantity to every venue's actual exposure or count parent and children as the same resting venue order. Aggregate venue-level bounds once per venue/account rather than once per member. + +This policy can reject a parent whose current split would fit limits but whose future full allocation to one member would not. That is an explicit tradeoff for review. An allocation-only admission policy would be less restrictive but would allow accepted parents to stall on later child risk checks. Actual child dispatch always rechecks current risk; admission is neither a funds reservation nor a guarantee of later acceptance. Existing ineffective or incomplete risk rules, such as a venue-notional rule without supplied exposure, must be identified in verification rather than described as enforced. + +## Persistence, query and compatibility policy + +Retain managed_orders as the authoritative routed-parent snapshot and retain the shared order projection for now. PRE-5's removal of parent copies is a separate issue, not a dependency to implement here. + +The shared projection's current `(venue,id)` primary key requires non-null values. Use an explicitly private reserved managed storage namespace for routed parent rows, while keeping their serialized public execution venue null. Centralize conversion between storage identity and execution identity so no caller accidentally treats the storage namespace as a venue. Never configure an adapter/account for that namespace. Parent lookup, including finished parents after restart, must use managed ID lookup rather than scanning only configured adapters. + +Persist a normalized parent execution-account membership relation, keyed by parent ID and account key with its concrete venue, alongside each new parent snapshot. Use it for indexed venue/account filtering and scope lookup without scanning and decoding every historical snapshot or fetching current mutable bucket definitions. A parent matches any eligible venue filter and appears once in an unfiltered list. Preserve stable pagination using the private storage key and parent ID; reject incompatible pre-upgrade cursors clearly and require clients to restart pagination. + +D3 accepted on 25 October 2026: save the parent snapshot, execution-account membership, shared readable order projection, submission-result bookkeeping and related journal events in one transaction. Refactor journal writes into transaction-owned operations and helpers that do not start or commit transactions themselves. Update in-memory caches only after a successful commit; propagate cancellation/failure without publishing uncommitted state. Reuse this transaction boundary for parent transitions coupled to child intents. Verify rollback and restart with failures injected between writes on both databases. + +Add an immutable Alembic revision for the relation and routed-parent projection rekey/backfill. Migrate routed managed snapshots, shared projections and account indexes consistently; preserve parent IDs, client IDs, user identity, statuses, fills, child IDs and actual child venue/account attribution. Preserve old nominal values as legacy provenance, not operative ownership. Do not edit old audit events to pretend they originally contained the new contract. + +Infer historical accounts only from unambiguous persisted child/intent evidence and the historical parent account for its matching venue. Never infer a previously unknown execution account from the current configuration. If a legacy live parent lacks sufficient evidence for an enabled venue, require an explicit reviewed mapping before enabling dispatch; do not silently broaden access or move execution. Test and document this case as an upgrade preflight requirement. The migration and startup validation must fail transactionally on conflicting evidence. + +Update HTTP response schemas and any experimental Python callers that assume a string parent venue. Scope predicates must be shared by list/detail/amend/cancel and parent-bearing event delivery/replay. The current event socket sends subscribed events without account-level filtering; parent scope filtering must cover both replay and live delivery, and this ticket must not claim that unrelated event types have thereby received a complete security audit. + +Require all-account visibility for a complete parent/report and all-account trade permission for a parent action. A caller may still see an authorized individual child through existing child routes without seeing the whole aggregate. Do not expose another venue's fills through a partial-account parent report. + +Upgrade both SQLite and PostgreSQL through the existing application-controlled migration path and one execution worker. Back up and verify restore first. Old binaries must not open the upgraded journal; downgrade requires a compatible backup and resolved outstanding venue activity. SDK merge precedes workspace merge, and workspace CI must verify the actual resulting experimental branch tip. + +## Verification plan + +Verification uses existing PaperVenue and HTTP test fixtures without exchange credentials. Actual results are recorded separately from this design in the linked workspace verification report. + +```text +Engine.submit(bucket request) / POST /orders on the standalone trading app + -> saved scope -> risk -> durable parent + -> existing router -> two scripted venue adapters + -> child responses/fills -> parent/report + -> list/detail/filter/cancel/halt + -> process restart + migrated database -> same identities/scope + +Terminal fixture book -> select bucket -> enter size/worst price + -> routed preview -> correct per-venue legs + -> contract checks + frontend build (no order submission UI) +``` + +1. **Ordering invariance:** run equivalent buckets with member order reversed and unambiguous best prices; assert identical owner, saved account map, risk/permission decisions and economic allocations, ignoring generated IDs and timestamps. Include all allocation on the non-first venue, multi-venue fills, disabled members and flipped YES/NO members. Assert actual requests delivered to each adapter carry that venue's exact named account. +2. **Risk and authorization:** deny a non-first venue/account, halt it, restrict its market, exceed member/account position limits, and exercise mixed-view/trade grants. Verify denial before dispatch where admission requires it, exact child-side checks after flipping, authorized child access, and no parent/report/live-event/replay leakage to a partial-account key. Check the proposed conservative admission tradeoff explicitly. +3. **Lifecycle and storage:** use resting children, partial fills, colliding venue-native order IDs, delayed cancellation confirmations and cancellation/fill races. Cancel or halt through the public entry points and assert every child is addressed on its own venue; parent stays pending until tracked children resolve. Restart working, cancelling and finished parents; verify ID lookup, reports, filters, pagination, saved account scope and unchanged aggregate fills. A changed bucket definition or account configuration must not silently alter the instruction. +4. **Upgrade:** prepare pre-change databases with live and finished routed parents, child evidence, incomplete/conflicting account evidence and ordinary single-venue orders. Apply the new migration on SQLite and PostgreSQL; verify identity/count/state preservation, no duplicate projection after rekey, explicit failure for unsafe mappings, and successful restart/cancel of migrated parents. Repeat normal startup to verify idempotence. Preserve known recovery limitations rather than claim a new all-children reconciliation barrier. +5. **Terminal compatibility:** extend the installed-consumer smoke check with a minimal offline bucket submission and scope/child-account assertions. Run terminal backend tests, generated-contract consistency and frontend build against the exact candidate SDK commit. In a browser with deterministic market fixtures, select a two-venue bucket, enter size and worst price, inspect routed preview legs, reverse member order and compare. Confirm Submit remains disabled as specified on main. +6. **Release evidence:** require SDK native and pure-Python tests, Rust checks and PostgreSQL storage CI, plus workspace terminal CI against the merged experimental branch tip. Record both tested commit hashes and sanitized results. Create linked PRs, run `/review`, resolve findings, merge SDK to `codex/order-lifecycle-storage` first, then reverify and merge prediction_market to `main`. + +## Implementation scope proposed for engineering review + +Expected SDK areas: trading models; engine bucket admission/child dispatch/risk integration; managed base/controller/routed scope; journal and projection lookup/filtering; membership schema and Alembic migration; server authorization/events; lifecycle documentation; routing/API/recovery/migration/storage tests. Workspace areas: installed-consumer smoke test, terminal integration decision/rollout documentation and any necessary CI compatibility wiring. This necessarily exceeds eight files but introduces no new service or execution worker. + +Suggested sequence: define the scope/model and migration contract; implement shared scope/identity helpers and storage; update admission/dispatch and risk; update server authorization/query/event paths; add regressions and consumer verification; run review and land in dependency order. Both worktrees remain isolated throughout. Cross-repository tasks are sequential where the consumer depends on the SDK contract. + +## Required verification report + +User requirement confirmed on 25 October 2026: deliver a verification report with screenshots and code snippets showing inputs and actual captured outputs. Do not make actual trades. Use offline scripted/PaperVenue adapters with no real exchange execution credentials; browser verification uses deterministic market fixtures. + +The final report must include: + +- Exact workspace and SDK commits tested, database backend and migration revision, commands or runnable snippets, test counts/results, and CI links where available. +- Minimal SDK and HTTP request snippets, paired with sanitized captured responses demonstrating a venue-neutral parent, its saved execution accounts, and the actual venue/account on each child. Clearly label all examples as simulated execution. +- Before/after or paired runs for reversed bucket members, all allocation to the non-first venue, and a flipped member, with expected versus observed outcomes. +- Evidence for permission/risk denials, venue-scoped halt and cancellation, restart/finished-parent lookup, and legacy migration. Show relevant persisted state or query results without exposing credentials or private account data. +- Screenshots of the terminal's deterministic two-venue bucket preview, including the entered quantity/worst price and resulting venue legs, plus readable captures of SDK/API verification results where useful. Label screenshots by scenario and distinguish preview evidence from simulated execution evidence; the current terminal cannot submit orders. +- A requirement-to-evidence table linking ticket acceptance criteria to tests, screenshots and captured outputs, with explicit pass/fail/not-run status and remaining limitations. Never substitute illustrative output or a mockup for observed results. + +Keep the report and its screenshot assets available as reviewable artifacts and link them from both PRs. Finish the report before requesting the final merge decision or merging under the user's conditional authorization. No live execution is part of this report. + +## Existing mechanisms to reuse and exclusions + +Reuse user identity, captured routable bucket, market-ID venue resolution, fee-aware allocation, per-child risk, parent IDs, managed snapshots, per-child venue cancellation, RoutedLimit.venues(), scripted venue fixtures and existing CI. Do not build a new router, order state machine, account service or terminal trading screen. + +Outside scope: public SDK main/release/PyPI, production dependency changes, live trading, PRE-5 projection removal, multi-user/multi-worker execution, a multi-account selection UI, routing tie-break redesign, and the broader uncertain-send/resync redesign documented in routed.md. No unrelated TODO cleanup. + +## Approved decisions + +D1 retains two linked repository PRs. D2 chooses full-capacity admission at every +eligible destination. D3 requires atomic parent/account/projection persistence. +D4 captures one scope for authorization, risk and execution. D5 requested an +independent review; D6 includes HTTP audit-route permissions and D7 includes +account checks before recovery/cleanup I/O. D8 explicitly approved implementation +with those additions. The user requested all remaining material questions in one +batch and prohibited actual trades. The verification report must include actual +screenshots, runnable input snippets, captured outputs and exact tested commits. + +## Engineering review: architecture + +Scope accepted in D1. Three findings were resolved with explicit user decisions: (1) D2 chooses conservative full-quantity admission at every eligible destination; (2) D3 makes parent state, scope, projection and related bookkeeping atomic; (3) D4 authorizes and executes the same captured snapshot. No new service, worker, dependency or published artifact is introduced. Existing SDK and terminal build/CI pipelines remain the distribution and compatibility checks. + +Migration is deliberately fail-closed for ambiguous historical account evidence. Preflight must identify such cases before upgrade while the old environment is still available; resolve mappings explicitly rather than starting new execution under guessed accounts. Account mismatches after restart block new dispatch rather than rewrite the saved scope. Scoped event delivery checks both replay and live delivery. These policies are already included above and require no additional architecture decision. + +## Engineering review: code quality + +No additional issues found against the revised plan. Reuse one scope preparation/validation implementation, one authorization predicate for complete parents, one private storage-identity conversion, and transaction-free low-level write helpers called by transaction-owning journal methods. The storage protocol must expose the atomic parent-save operation consistently for both backends; PostgreSQL already inherits the shared journal operations. + +Do not spread `venue is None` fallbacks across adapters or mint a synthetic Account. Model validation distinguishes a cross-venue engine parent from a venue-held order. Keep existing single-venue behavior on its existing path. Storage and lifecycle diagrams must describe the actual transaction boundary and distinguish attempted cancellation/resync from confirmed venue outcomes. Prior cancellation/restart fixes in commits `e2df3ba` and merge `27b4ea4` identify an area for regression testing; their state mappings and delayed-confirmation behavior are retained. + +## Engineering review: tests and failure modes + +Frameworks: SDK uses pytest/anyio with PaperVenue and FastAPI TestClient; SQLite/PostgreSQL contract tests already share scenarios. Terminal uses unittest backend tests plus generated-contract checks and TypeScript/Vite build; browser checks cover observable preview behavior. Existing routed tests cover fee-driven non-first allocation, flips, cross-venue cancellation, scoped halts and delayed-cancellation recovery. Extend these fixtures rather than invent another simulation engine. + +The original verification plan plus the accepted D3/D4 regression requirements covers the planned paths. No additional test-policy decision is needed. The following map specifies branches and expected observable outcomes; implementation must add assertions to the named suites and record actual results in the report. + +```text +submit: trusted prepare scope [test_routed_limit / test_serve_buckets] + |-- missing/inactive/empty bucket, missing adapter -> explicit refusal, no send + |-- singular account or mismatched account/venue -> explicit refusal, no send + |-- valid -> capture enabled members/accounts exactly once + |-- concurrent bucket edit -> captured scope remains operative [D4] + |-- unauthorized account -> 403, no parent dispatch + |-- global/member halt or member risk denial -> rejection, no dispatch + |-- all checks pass -> atomic save [test_order_storage] + |-- failure/cancellation between any writes -> full rollback [D3] + |-- commit -> update memory and start existing router + |-- normal/flipped/split/non-first allocation -> exact child accounts + |-- child risk changes -> existing child rejection behavior + |-- fills/status repeats -> unchanged cumulative attribution + +read/action: parent ID [test_server_trading / test_serve_buckets] + |-- partial grants -> hide/refuse complete parent and report + |-- all grants -> list/detail/report/cancel/amend as supported + |-- venue/account filter -> scope membership, one parent row, stable paging + |-- live/replayed parent event -> same visibility rule + |-- unknown/malformed identity or cursor -> explicit error, no fallback account + +restart/upgrade [test_routed_recovery / test_owner_migrations / test_postgres_storage] + |-- fresh DB -> new schema; repeated start -> unchanged state + |-- legacy, unambiguous -> preserve IDs/children/fills, rekey parent once + |-- missing/conflicting historical account -> explicit preflight/migration failure + |-- failure/cancellation during migration -> unchanged old data, reopen safely + |-- unfinished parent -> restore scope, reconcile/cancel original children + |-- finished parent -> ID lookup/report still works without live controller entry + |-- changed account config -> block new dispatch, retain original attribution + |-- same client-ID retry where supported -> original parent/scope, no duplicate send + +terminal [backend unittest / consumer smoke / browser] + fixture books -> select bucket -> enter size/worst price -> preview venue legs + |-- reversed members / flipped outcome -> expected equivalent economics + |-- missing liquidity -> visible remainder + |-- Submit -> remains disabled (preview only) +``` + +| Failure mode | Planned error handling and user-visible result | Required evidence | +|---|---|---| +| Bucket changes during permission check | Continue using the authorized captured snapshot | Interleaved edit test, assert adapters receive only captured members | +| Parent save fails after membership/projection write | Roll back the whole transaction; no accepted parent/event/cache update | Inject failure at each boundary on SQLite and PostgreSQL, reopen and compare all rows | +| Historical account evidence is missing or conflicts | Refuse unsafe migration/dispatch with affected parent diagnostic | Legacy fixtures and rollback assertions on both databases | +| Account configuration changes after restart | Refuse new sends through a different account | Restart fixture, zero new sends, unchanged saved identity | +| Partial-account caller requests parent or replays events | Hide aggregate or return permission error; no unauthorized account data | HTTP/report/live-event/replay grant matrix | +| Two venues return the same native order ID | Match children by concrete venue plus native ID | Two-venue cancellation/fill fixture | +| Cancel acknowledgement remains open or races with fill | Retain pending cleanup and cumulative fills; never claim unconfirmed cancellation | Existing cancellation scenarios extended to null-venue parents and restart | +| Parent is finished before restart | Retrieve by managed ID without scanning adapters | Finished-parent detail, cancel/idempotent read and report tests | +| Many parents or repeated scope lookups | Indexed membership filters and batched scope retrieval | Query-count and query-plan assertions; no full-history snapshot scan | + +No unaddressed silent critical failure mode was identified in this plan review. The broader uncertain-send/resync limitations remain explicitly outside this change; this is not a claim of production trading readiness or a test result. + +## Engineering review: performance + +No additional issues found against the planned normalized membership relation. Index concrete venue/account membership for filtered queries, use `EXISTS` or deduplicated joins so a parent is returned once, and load page scopes in a batch rather than one lookup per row. The snapshot remains authoritative; membership is its transactional query projection, never an independent mutable account-selection source. + +Parent admission scales with enabled members, not historical parent count. Reuse one view of the engine's current open orders per admission and group by venue/account before per-member checks; actual children retain their existing checks. Keep order/fee planning unchanged. Test query counts with several page sizes and inspect SQLite and PostgreSQL query plans for the membership filters. Do not add speculative caching or a benchmark service. + +## Implementation Tasks + +- [x] **T1 (P2)** — SDK scope/admission — implement the venue-neutral routed-parent contract and the D2 full-capacity risk policy; reuse concrete-account child dispatch. Verify reversed members, all allocation to non-first venue, flips, named accounts and conservative rejection of an otherwise valid split. +- [x] **T2 (P1)** — SDK journal/migrations — implement D3 atomic parent persistence and scope membership, private projection identity and legacy backfill. Verify injected failures, cancellation, restart and migration on both backends. +- [x] **T3 (P1)** — SDK submission/server — implement D4 single captured scope and consistent scope-based read/action/event authorization. Verify the interleaved bucket edit and the all-account/partial-account permission matrix. + +These tasks derive from the three findings above; ordinary implementation steps and the required verification report remain described in their own sections. Human/agent effort estimates are not supplied because this repository has no applicable compression table; measure actual work rather than invent precision. No new tasks arose from code quality, test-policy or performance review. + +## Dependencies and worktrees + +| Step | Modules | Depends on | +|---|---|---| +| Contract and captured scope | SDK trading and engine | — | +| Atomic storage and migration | SDK engine storage and migrations | Contract | +| Submission, risk, authorization and recovery | SDK engine and server | Contract, storage | +| SDK regression verification | SDK tests | Submission/storage paths | +| Consumer check and terminal evidence | Workspace research, terminal and CI | Candidate SDK | +| Review and landing | Both PRs | All checks and report | + +Sequential implementation in two isolated repository worktrees: the consumer depends on the SDK contract and no parallel agent work is required. SDK regression and terminal checks may run independently once the candidate SDK is stable. Merge SDK into the experimental branch, then rerun workspace checks against that resulting branch commit before merging workspace main. + +## NOT in scope + +- Public SDK main, release/PyPI and production pins: the intended dependency remains the experimental branch. +- Actual trades: the user explicitly prohibited them; all execution evidence is simulated. +- Terminal trading UI: main only previews orders, and enabling execution is a separate product change. +- PRE-5 projection removal: retaining the projection avoids coupling two storage redesigns. +- Multi-user, multi-worker and account-selection UI: reuse current ownership and configured adapters. +- Routing ties and broader uncertain-send/resync redesign: preserve the existing behavior and documented limitations. + +## What already exists + +Reuse the user owner registry, bucket routable snapshot, fee-aware router, venue-specific child dispatch, child-based cancellation, `RoutedLimit.venues()`, managed snapshot recovery, Alembic upgrades and shared SQLite/PostgreSQL test scenarios. Reuse terminal CI's SDK installation, backend tests, consumer smoke check, schema check and frontend build. No existing TODO file was found; no new deferred TODO is proposed. Existing lifecycle follow-ups remain documented in routed.md and routed-lifecycle-plan.md. + +## Independent review resolutions + +The fresh-context independent review found two gaps, both accepted by the user: +HTTP `/journal` and `/order-log` now fall under the saved parent-scope access +contract, and account identity checks must precede early recovery reads and +cleanup writes as well as new dispatch. Legacy events resolve durable scope; +unknown parent attribution is withheld from partial-account viewers. A changed +account must remain unresolved rather than be marked lost/final by reading a +replacement account. These additions are part of the approved scope. + +## GSTACK REVIEW REPORT + +| Review | Trigger | Why | Runs | Status | Findings | +|---|---|---|---|---|---| +| CEO Review | — | Scope strategy | 0 | Not needed | Contract correction | +| Independent plan review | Outside voice | Fresh-context challenge | 1 | Resolved | 2 P1 findings accepted in D6/D7 | +| Eng Review | `/plan-eng-review` | Architecture, code quality, tests, performance | 1 | Clear | 5 findings resolved through explicit decisions | +| Design Review | — | Visual experience | 0 | Not needed | No UI redesign | +| DX Review | — | Developer experience | 0 | Not needed | Existing setup/CI | + +- **UNRESOLVED:** 0 plan decisions. +- **VERDICT:** Engineering plan cleared and implementation approved. Code review, offline verification report and satisfactory CI still gate landing. No actual trades are authorized. diff --git a/synpath/engine/orders/lifecycle/nominal-venue-verification-plan.md b/synpath/engine/orders/lifecycle/nominal-venue-verification-plan.md new file mode 100644 index 0000000..c84a7f0 --- /dev/null +++ b/synpath/engine/orders/lifecycle/nominal-venue-verification-plan.md @@ -0,0 +1,63 @@ +# PRE-6 verification plan + +Prepared during `/plan-eng-review`, 25 October 2026. Status: approved and executed offline; final CI and merged-tip evidence gate landing. Full design: [nominal-venue-removal-design.md](nominal-venue-removal-design.md). Implementation was explicitly approved with audit-route and recovery safeguards. All execution is offline through scripted venues; do not use real trading credentials or place actual trades. The linked workspace report is `research/synpath-public-testing/PRE6_VERIFICATION.md`; it records observed results and limitations. + +## Affected pages and routes + +- Standalone SDK trading app: `POST /orders`, `GET /orders`, `GET /orders/{id}`, `PATCH /orders/{id}` where supported, and `DELETE /orders/{id}`. Confirm mounted prefixes for any other server configuration rather than assuming `/trading`. +- Bucket report/preview routes: use the existing router's paths and test fixtures; assert complete-parent visibility across all captured accounts. +- SDK event socket and replay: verify parent and report events under complete and partial grants. +- Terminal at `http://127.0.0.1:5173`: select bucket, execution ticket and routed preview. Use deterministic market-data fixtures. Order submission remains disabled. + +## Key interactions to verify + +1. Submit a bucket with two named accounts through `Engine.submit()` and through HTTP. Observe null parent execution venue/account, stable user identity and a saved map of eligible execution accounts. Verify exact concrete venue/account on requests received by each scripted adapter. +2. Reverse otherwise identical member order with distinct prices. Compare owners, account maps, permissions, risk decisions and economic allocation, excluding generated IDs/timestamps. +3. Make the non-first venue strictly cheaper and deep enough for the whole order. Observe all fills there while parent execution venue/account remain null. +4. Flip a member. Verify its actual outcome, side and YES price are converted correctly; parent quantities/prices remain in bucket terms. +5. Revoke one member-account grant, halt a non-first venue, and exceed its position limit in separate scenarios. Observe denial according to the agreed full-capacity admission policy before dispatch. +6. Interleave a bucket edit after permission checks. Observe execution only on the originally authorized captured members/accounts. +7. Submit resting children, cancel the parent, delay one cancellation acknowledgement, then deliver a final status or fill. Observe pending cleanup until children resolve and accurate aggregate fills. +8. Restart working, cancelling and finished parents. Observe original scope/IDs, lookup and reports; account configuration changes never redirect new sends. +9. On the terminal, enter a quantity and worst price for deterministic two-venue books. Observe correct routed preview legs and remainder; reverse members and repeat. Capture screenshots with the entered fields and resulting legs visible. + +## Edge cases + +- Missing/inactive bucket, no enabled members, missing adapter, mismatched account venue and caller-supplied singular account. +- Partial view/trade grants: refuse or hide complete aggregate/report and scoped events, while preserving authorized individual-child access. +- Same venue-native child ID on two distinct venues, duplicate child status messages, partial/full fills during cancellation. +- Venue/account filters and pagination across live and finished parents: return each matching parent once, use saved membership, and handle incompatible old cursors explicitly. +- Conservative capacity example: order 100 with capacity 60 on each venue is rejected even though a 50/50 split would fit. Label this intentional policy in the report. +- Missing/conflicting legacy account evidence: explicit preflight/migration failure, no guessed current account. +- Same-client-ID retry where supported: original identity/scope and no duplicate send. Document any unsupported existing bucket retry behavior rather than claim an untested guarantee. + +## Critical paths + +| Path | Test suite / evidence | Required result | +|---|---|---| +| SDK submit to actual child requests | `tests/test_routed_limit.py` and captured request/response snippets | Exact venue accounts; member-order invariance; flipped correctness | +| HTTP permissions to dispatch | `tests/test_server_trading.py`, `tests/test_serve_buckets.py` | No unauthorized sends or parent aggregate leakage | +| Captured snapshot under concurrent edit | HTTP/engine interleaving fixture | Authorized snapshot equals executed snapshot | +| Atomic parent save | `tests/test_order_storage.py` plus portable PostgreSQL scenarios | Inject failure/cancellation between each write; all rows/events roll back | +| Legacy migration | `tests/test_owner_migrations.py` on SQLite/PostgreSQL | Preserve identities/fills, one rekey, unsafe mapping fails transactionally | +| Cancellation and restart | `tests/test_routed_recovery.py` and existing pending-cancellation regressions | No premature completion or new allocation while cancelling | +| Query scaling | Membership-filter query plans and page query counts | Indexed membership and bounded queries per page | +| Terminal integration | Backend unittest, consumer smoke, contract check, frontend build | Compatible installed candidate SDK and unchanged preview-only terminal | +| Browser evidence | Deterministic bucket screenshots | Visible inputs, venue legs and remainder; no simulated order presented as a live trade | + +## Evidence and commands + +Run SDK targeted tests first, then the existing offline SDK suite with its live/demo exclusions, pure-Python variant and required native/Rust checks. PostgreSQL scenarios require a disposable database through `SYNPATH_TEST_POSTGRES_DSN`; a skipped database suite is not a pass. Require CI's PostgreSQL job or equivalent actual local results. + +Workspace checks use its existing commands: + +```sh +uv run --project terminal --locked python -m unittest discover -s terminal/backend/tests -v +uv run --project terminal --locked python research/synpath-public-testing/check_install.py +npm --prefix terminal/frontend run check:contracts +npm --prefix terminal/frontend run build +``` + +Candidate SDK worktrees are deliberately on a feature branch. The setup script currently requires a clean clone on the experimental branch and `.git` to be a directory. Use a documented isolated candidate-install arrangement for pre-merge tests; do not reset the original checkout, claim that setup verified a feature worktree, or add a general setup bypass. After merging the SDK PR, run the normal workspace setup and CI against the resulting experimental branch tip and record its actual hash. + +For every scenario, save the exact command or short runnable input snippet, expected result, captured sanitized output, pass/fail/not-run status and limitation. Use fixture account names only. Screenshots must be actual captures, linked by scenario; no mockup may stand in for observed behavior. Record SDK/workspace hashes, schema revision, test totals and CI links. Report actual child fills as simulated, never real. Link the final requirement-to-evidence report and screenshot assets from both PRs before final landing. diff --git a/synpath/engine/orders/lifecycle/routed.md b/synpath/engine/orders/lifecycle/routed.md index 54aad0c..c52a5c0 100644 --- a/synpath/engine/orders/lifecycle/routed.md +++ b/synpath/engine/orders/lifecycle/routed.md @@ -8,6 +8,24 @@ Status: current behavior, not the proposed contract. Read alongside [shared mana [`RoutedLimit`](../routed.py) manages this instruction despite its historical class name. Its optional settings include `min_stay_s`, `max_rounds`, `max_age_s` and precision overrides. Zero routing limits mean disabled bounds. Size and price validation also comes from `OrderRequest` and risk checks. These are parent instructions; children are priced GTC limit orders. +## Parent identity and execution accounts + +The parent has no execution venue or singular account. Its stable user owns the +instruction; `execution_accounts` captures exact accounts for the enabled members. +Each child selects its concrete venue/account from that saved scope. Member order +cannot change ownership, risk admission or permissions. Admission conservatively +checks full quantity at each eligible member and actual children are checked again. +Authorization and execution use one captured snapshot. Single-account bucket +requests are rejected. See [engine upgrade instructions](../../README.md#cross-venue-parent-identity-and-upgrade-preflight) +for migration, legacy account mappings and compatibility details. + +Parent snapshot, account membership, readable record and journal events commit +atomically. Parent lookup uses managed identity even after completion/restart; +actual venue orders retain their existing keys. Scope guards protect parent detail, +reports, actions, HTTP audit history and live/replayed events. Account mismatch +blocks affected recovery reads and cleanup as well as new dispatch, leaving the +activity unresolved until its original account is available. + ## Current transitions | Observation or action | Current transition and behavior | @@ -58,4 +76,4 @@ Offline regression tests in `tests/test_routed_limit.py: TestPendingCancellation ## Next phase boundary -The Engine owns the instruction; venues own child acceptance and fills. Routing progress and the intended pending ending use the existing managed snapshot path; no schema or table layout changed. The following phase must assess saved child identities, failed venue reads, uncertain sends and cancellation races across process failure before stronger recovery guarantees can be claimed. +The Engine owns the instruction; venues own child acceptance and fills. Routing progress and the intended pending ending use the managed snapshot path. PRE-6 adds account membership and rekeys parent projections through migration 0006; it does not change lifecycle states. The following phase must assess saved child identities, failed venue reads, uncertain sends and cancellation races across process failure before stronger recovery guarantees can be claimed. diff --git a/synpath/engine/orders/manager.py b/synpath/engine/orders/manager.py index 9df1a09..5ce5a91 100644 --- a/synpath/engine/orders/manager.py +++ b/synpath/engine/orders/manager.py @@ -25,6 +25,7 @@ from typing import Any, Iterable, Mapping, Type from ...errors import NetworkError +from ...trading.errors import SubmissionPending from ..storage_types import IntentState from ...trading.types import Fill, Order, OrderRequest, OrderType from .base import Child, Context, ManagedOrder, State, now_ms @@ -77,7 +78,8 @@ def watching(self, market_id: str) -> list[ManagedOrder]: # -- creating ------------------------------------------------------------- - def build(self, request: OrderRequest, *, venue: str, account: Any) -> ManagedOrder: + def build(self, request: OrderRequest, *, venue: str | None, account: Any, + execution_accounts: dict | None = None) -> ManagedOrder: """Construct the requested order implementation without saving or starting it.""" # `managed_kind` lets one venue type map to an engine-held variant: # a market order the engine walks the book with, rather than the @@ -89,9 +91,10 @@ def build(self, request: OrderRequest, *, venue: str, account: Any) -> ManagedOr f"{kind} is an engine-held order type with no implementation loaded; " f"known: {sorted(REGISTRY)}" ) - return cls(request, venue=venue, account=account) + return cls(request, venue=venue, account=account, execution_accounts=execution_accounts) - async def create(self, request: OrderRequest, *, venue: str, account: Any, owner: ManagedOrder | None = None) -> ManagedOrder: + async def create(self, request: OrderRequest, *, venue: str | None, account: Any, + execution_accounts: dict | None = None, owner: ManagedOrder | None = None) -> ManagedOrder: """Save and register an engine-held order, announce it, then start its behavior. `owner` is another managed order, not a user: for example, a bracket @@ -100,7 +103,7 @@ async def create(self, request: OrderRequest, *, venue: str, account: Any, owner """ # Choose the implementation (for example, RoutedLimit for a bucket). # A managed order can itself be a child of another managed order. - parent = self.build(request, venue=venue, account=account) + parent = self.build(request, venue=venue, account=account, execution_accounts=execution_accounts) parent.owner_id = owner.id if owner else None # Persist the initial instruction before start() can send venue orders. @@ -157,7 +160,12 @@ async def _run_hook(self, parent: ManagedOrder, call: Any) -> None: """Run one hook, catching what it raises so the others keep working.""" ctx = self.context(parent) try: + if parent.execution_accounts: + for venue, account in parent.execution_accounts.items(): + self.engine.assert_execution_account(venue, account) await call(ctx) + except SubmissionPending as exc: + parent.detail = str(exc) except (NetworkError, asyncio.TimeoutError): parent.detail = "venue action is unresolved; waiting for reconciliation" except Exception as exc: @@ -277,8 +285,7 @@ async def save(self, parent: ManagedOrder) -> None: for child in parent.children: self.by_child.setdefault(child.order_id, parent.id) self.by_child.setdefault(f"{child.venue}:{child.order_id}", parent.id) - await self.engine.journal.save_managed(parent.snapshot()) - await self.engine.journal.upsert_order(parent.as_order(), event=f"managed.{parent.state}") + await self.engine.journal.save_managed(parent.snapshot(), order=parent.as_order()) self.engine._remember(parent.as_order()) async def restore(self) -> int: diff --git a/synpath/engine/orders/routed.py b/synpath/engine/orders/routed.py index f1e0ca7..b1c6502 100644 --- a/synpath/engine/orders/routed.py +++ b/synpath/engine/orders/routed.py @@ -184,6 +184,8 @@ async def resync(self, ctx: Context) -> None: if adapter is None: continue try: + if self.execution_accounts: + ctx.engine.assert_execution_account(child.venue, self.execution_accounts[child.venue]) order = await adapter.fetch_order(child.order_id) except Exception as exc: await ctx.publish("managed.resync_failed", {"order_id": child.order_id, "venue": child.venue, diff --git a/synpath/engine/reconcile.py b/synpath/engine/reconcile.py index 0bdbaac..8b5eb97 100644 --- a/synpath/engine/reconcile.py +++ b/synpath/engine/reconcile.py @@ -114,6 +114,7 @@ async def run(self, venue: str | None = None) -> list[ReconcileReport]: async def venue(self, venue: str, adapter: TradingExchange) -> ReconcileReport: report = ReconcileReport(venue=venue) + await self.engine.guard_routed_venue_accounts(venue) await self._orders(venue, adapter, report) await self._fills(venue, adapter, report) await self._positions(venue, adapter, report) diff --git a/synpath/engine/risk.py b/synpath/engine/risk.py index 1d3b25c..ae2401c 100644 --- a/synpath/engine/risk.py +++ b/synpath/engine/risk.py @@ -216,14 +216,15 @@ def evaluate_order_risk( self, request: OrderRequest, *, - venue: str, - account: Account, + venue: str | None, + account: Account | None, open_orders: Iterable[Order] = (), risk_reference_price: Decimal | None = None, market_close_ts: int | None = None, event_id: str | None = None, venue_notional: Decimal | None = None, caller_supplied_order_id: bool = False, + instruction_only: bool = False, ) -> Decision: """Everything that must be true before this order is signed. @@ -290,6 +291,12 @@ def evaluate_order_risk( if len(self._sent) >= config.max_orders_per_minute: return self._no("order_rate", f"{len(self._sent)} orders in the last minute, at the {config.max_orders_per_minute} limit") + if instruction_only: + if config.daily_loss_limit is not None and self.ledger is not None: + if -self.realized_today() >= config.daily_loss_limit: + return self._no("daily_loss", "daily realized loss limit reached") + return self._ok() + if config.closing_soon_s is not None and market_close_ts is not None and request.leg[1] != "sell": left = (market_close_ts - now * 1000) / 1000 if left <= config.closing_soon_s: diff --git a/synpath/engine/storage.py b/synpath/engine/storage.py index eba0e1d..ed7730b 100644 --- a/synpath/engine/storage.py +++ b/synpath/engine/storage.py @@ -57,14 +57,14 @@ async def submission(self, client_order_id: str) -> dict[str, Any] | None: ... async def record_submission(self, original: OrderRequest, effective: OrderRequest, - *, venue: str, account: Account) -> None: + *, venue: str, account: Account | None) -> None: ... async def reject_submission(self, client_order_id: str, error: dict[str, Any]) -> None: ... async def plan_managed_child(self, snapshot: dict[str, Any], request: OrderRequest, - *, venue: str, account: Account) -> None: + *, venue: str, account: Account, parent_order: Order | None = None) -> None: ... async def record_intent(self, request: OrderRequest, *, client_order_id: str, venue: str, account: Account, operation: str='create', target_order_id: str | None=None) -> Intent: @@ -88,7 +88,7 @@ async def upsert_order(self, order: Order, *, event: str='order.updated') -> Non async def record_fill(self, fill: Fill, *, book: str | None=None, trader: str | None=None) -> bool: ... - async def order(self, venue: str, order_id: str) -> Order | None: + async def order(self, venue: str | None, order_id: str) -> Order | None: ... async def order_by_client_id(self, client_order_id: str) -> Order | None: @@ -128,7 +128,7 @@ async def buckets(self, *, book: str | None=None, status: str | None='active') - async def archive_bucket(self, bucket_id: str) -> bool: ... - async def save_managed(self, snapshot: dict[str, Any]) -> None: + async def save_managed(self, snapshot: dict[str, Any], *, order: Order | None = None) -> None: ... async def managed(self, *, states: Iterable[str] | None=None) -> list[dict[str, Any]]: @@ -137,6 +137,9 @@ async def managed(self, *, states: Iterable[str] | None=None) -> list[dict[str, async def managed_on(self, market_id: str) -> list[tuple[str, dict[str, Any]]]: ... + async def routed_account_keys(self, venue: str) -> list[str]: + ... + async def managed_one(self, parent_id: str) -> tuple[str, dict[str, Any]] | None: ... diff --git a/synpath/server/order_access.py b/synpath/server/order_access.py new file mode 100644 index 0000000..13d7aea --- /dev/null +++ b/synpath/server/order_access.py @@ -0,0 +1,73 @@ +"""Shared account authorization for order responses and their audit events.""" +from __future__ import annotations + +from ..engine.execution_scope import MANAGED_NAMESPACE, order_accounts +from ..trading.types import Account, Order + + +def may_view_account(principal, account: str) -> bool: + return principal.may("view", account) or principal.may("trade", account) + + +def may_view_order(principal, order: Order) -> bool: + return all(may_view_account(principal, account) for account in order_accounts(order)) + + +def may_trade_order(principal, order: Order) -> bool: + return all(principal.may("trade", account) for account in order_accounts(order)) + + +async def may_view_event(principal, engine, event) -> bool: + """Resolve legacy parent events through durable identity, never nominal account. + + New snapshots carry scope inline. Old append-only events are unchanged; + when their parent cannot be identified safely, only a global viewer sees + them. Individual child events remain visible on their concrete account. + """ + body = event.payload + request = body.get("effective_request") or body.get("request") or {} + if event.kind.startswith("submission."): + reservation = await engine.journal.submission(event.key) if event.key else None + if reservation and reservation["venue"] == MANAGED_NAMESPACE: + captured = reservation["effective_request"]["params"]["execution_accounts"] + return all(may_view_account(principal, Account.model_validate(account).key) for account in captured.values()) + is_parent = event.kind.startswith("managed.") or body.get("held_by") == "engine" + # Request params are caller-controlled, even when recorded in an audit + # event. Only engine-produced order snapshots or stored reservations own scope. + scope = body.get("execution_accounts") + if scope and is_parent: + return all(may_view_account(principal, Account.model_validate(account).key) for account in scope.values()) + if is_parent: + parent_id = body.get("parent_id") or body.get("id") or event.key + if parent_id and parent_id.startswith(f"{MANAGED_NAMESPACE}:"): + parent_id = parent_id.split(":", 1)[1] + stored = await engine.journal.managed_one(parent_id) if parent_id else None + if stored: + parent = engine._rebuilt(*stored) + return parent is not None and may_view_order(principal, parent.as_order()) + return principal.may("view", "*") + account = body.get("account") or request.get("account") + if isinstance(account, dict): + return may_view_account(principal, Account.model_validate(account).key) + if isinstance(account, str) and account != MANAGED_NAMESPACE: + return may_view_account(principal, account) + if scope: + return all(may_view_account(principal, Account.model_validate(account).key) for account in scope.values()) + parent_id = body.get("parent_id") or (request.get("tags") or {}).get("parent") + if parent_id: + stored = await engine.journal.managed_one(parent_id) + if stored: + parent = engine._rebuilt(*stored) + return parent is not None and may_view_order(principal, parent.as_order()) + return principal.may("view", "*") + if event.kind.startswith(("order.", "fill.", "intent.", "submission.", "risk.")): + # Legacy order events sometimes name only their client/native ID. + key = body.get("client_order_id") or event.key + order = await engine.journal.order_by_client_id(key) if key else None + if order is None and key: + try: + order = await engine._known(key.split(":", 1)[-1], None) + except Exception: + pass + return may_view_order(principal, order) if order else principal.may("view", "*") + return True diff --git a/synpath/server/trading.py b/synpath/server/trading.py index 70fb8af..bc3a627 100644 --- a/synpath/server/trading.py +++ b/synpath/server/trading.py @@ -33,6 +33,9 @@ import base64 import json import logging +from ..engine.execution_scope import BucketNotFound +from .order_access import may_view_order, may_trade_order, may_view_event + from decimal import Decimal from typing import Annotated, Any, Iterable, Literal @@ -68,6 +71,7 @@ SubmissionPending: 202, CancelUnconfirmed: 409, OrderNotFound: 404, + BucketNotFound: 404, InsufficientFunds: 400, OrderRejected: 400, InvalidOrder: 400, @@ -102,16 +106,19 @@ class first. A venue refusing our own credentials is 502: the caller's token def orders_cursor(key: tuple[int, str, str]) -> str: """The paging key of a page's last order, `(created_ts, venue, id)`, as an opaque token.""" - return base64.urlsafe_b64encode(json.dumps(list(key)).encode()).decode().rstrip("=") + return base64.urlsafe_b64encode(json.dumps({"v": 2, "key": list(key)}).encode()).decode().rstrip("=") def orders_cursor_key(cursor: str | None) -> tuple[int, str, str] | None: if not cursor: return None try: - created_ts, venue, order_id = json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4))) + decoded = json.loads(base64.urlsafe_b64decode(cursor + "=" * (-len(cursor) % 4))) + if not isinstance(decoded, dict) or decoded.get("v") != 2: + raise ValueError("old cursor; restart pagination") + created_ts, venue, order_id = decoded["key"] return int(created_ts), str(venue), str(order_id) - except (ValueError, TypeError): + except (ValueError, TypeError, KeyError): raise HTTPException(status_code=400, detail="cursor is not one this server gave out") from None @@ -539,14 +546,10 @@ async def status(principal: Caller, engine: EngineDep) -> EngineStatus: status_code=201) async def create_order(principal: Caller, engine: EngineDep, body: OrderRequest) -> Order: if is_bucket_id(body.market_id): - # A bucket order puts legs on every member venue: the caller must - # be allowed to trade on each of them. - row = await engine.journal.bucket(bucket_id_of(body.market_id)) - if row is None: - raise HTTPException(status_code=404, detail=f"{body.market_id}: no such bucket") - for member_venue in sorted(Bucket.model_validate(row).venues()): - require(principal, "trade", account_key(None, engine, member_venue)) - venue = None + def authorize(accounts): + for account in accounts.values(): + require(principal, "trade", account.key) + return await engine.submit(body, authorize_execution_accounts=authorize) else: venue = order_venue(body, engine) if venue is None: @@ -662,18 +665,20 @@ async def bucket_position( responses=RESPONSES) async def bucket_orders(principal: Caller, engine: EngineDep, bucket_id: Annotated[str, Path()]) -> PageResponse[BucketOrderReport]: - bucket = await _bucket(principal, engine, bucket_id) - reports = [bucket_report(p) for p in await engine.bucket_orders(bucket.id)] + if await engine.journal.bucket(bucket_id) is None: + raise HTTPException(status_code=404, detail=f"no bucket {bucket_id}") + reports = [bucket_report(p) for p in await engine.bucket_orders(bucket_id) if may_view_order(principal, p.as_order())] return PageResponse[BucketOrderReport](data=reports, next_cursor=None, count=len(reports)) @router.get("/buckets/{bucket_id}/orders/{order_id}", tags=["buckets"], summary="One order on a bucket", responses=RESPONSES) async def bucket_order(principal: Caller, engine: EngineDep, bucket_id: Annotated[str, Path()], order_id: Annotated[str, Path()]) -> BucketOrderReport: - bucket = await _bucket(principal, engine, bucket_id) parent = await engine.bucket_order(order_id) - if parent is None or parent.market_id != bucket.market_id: + if parent is None or parent.market_id != f"bucket:{bucket_id}": raise HTTPException(status_code=404, detail=f"no order {order_id} on bucket {bucket_id}") + if not may_view_order(principal, parent.as_order()): + raise HTTPException(status_code=403, detail="this key may not view every execution account") return bucket_report(parent) @router.get("/orders", tags=["orders"], summary="Orders: open ones, or every state", responses=RESPONSES) @@ -704,32 +709,23 @@ async def list_orders( open_only=scope == "open", limit=limit) orders = [current_view(engine, order) for order, _ in rows if order.user_id in (None, engine.config.user_id) - and visible(principal, (order.account.key if order.account else f"{order.venue}:default"))] + and may_view_order(principal, order)] next_cursor = orders_cursor(rows[-1][1]) if limit is not None and len(rows) == limit else None return PageResponse[Order](data=orders, next_cursor=next_cursor, count=len(orders)) @router.get("/orders/{order_id}", tags=["orders"], summary="One order", responses=RESPONSES) async def get_order(principal: Caller, engine: EngineDep, order_id: Annotated[str, Path()]) -> Order: - parent = engine.managed_order_controller.get(order_id) - order = parent.as_order() if parent is not None else None - if order is None: - for venue in engine.adapters: - order = await engine.journal.order(venue, order_id) - if order is not None: - break - if order is None: - raise HTTPException(status_code=404, detail=f"no order {order_id}") - account = order.account.key if order.account else f"{order.venue}:default" - if not visible(principal, account): - raise HTTPException(status_code=403, detail="this key may not view that account") + order = await _known(engine, order_id) + if not may_view_order(principal, order): + raise HTTPException(status_code=403, detail="this key may not view every execution account") return order @router.patch("/orders/{order_id}", tags=["orders"], summary="Amend an order", responses=RESPONSES) async def edit_order(principal: Caller, engine: EngineDep, order_id: Annotated[str, Path()], body: EditBody) -> Order: order = await _known(engine, order_id) - account = order.account.key if order.account else f"{order.venue}:default" - require(principal, "trade", account) + if not may_trade_order(principal, order): + raise HTTPException(status_code=403, detail="this key may not trade on every execution account") try: return await engine.edit(EditRequest(order_id=order_id, **body.model_dump(exclude_none=True))) except SynpathError as exc: @@ -741,8 +737,8 @@ async def edit_order(principal: Caller, engine: EngineDep, order_id: Annotated[s @router.delete("/orders/{order_id}", tags=["orders"], summary="Cancel an order", responses=RESPONSES) async def cancel_order(principal: Caller, engine: EngineDep, order_id: Annotated[str, Path()]) -> Order: order = await _known(engine, order_id) - account = order.account.key if order.account else f"{order.venue}:default" - require(principal, "trade", account) + if not may_trade_order(principal, order): + raise HTTPException(status_code=403, detail="this key may not trade on every execution account") try: return await engine.cancel(order_id) except SynpathError as exc: @@ -750,14 +746,10 @@ async def cancel_order(principal: Caller, engine: EngineDep, order_id: Annotated raise async def _known(engine: Engine, order_id: str) -> Order: - parent = engine.managed_order_controller.get(order_id) - if parent is not None: - return parent.as_order() - for venue in engine.adapters: - order = await engine.journal.order(venue, order_id) - if order is not None: - return order - raise HTTPException(status_code=404, detail=f"no order {order_id}") + try: + return await engine._known(order_id, None) + except OrderNotFound: + raise HTTPException(status_code=404, detail=f"no order {order_id}") from None # -- fills, positions, balances ------------------------------------------ @@ -825,7 +817,13 @@ async def get_order_log( repeats folded. Read only.""" require(principal, "view") from ..engine.order_log import order_log - rows, cursor = await order_log(engine, order_id=order_id, limit=limit, before=before) + if order_id: + order = await _known(engine, order_id) + if not may_view_order(principal, order): + raise HTTPException(status_code=403, detail="this key may not view every execution account") + async def allow(event): + return await may_view_event(principal, engine, event) + rows, cursor = await order_log(engine, order_id=order_id, limit=limit, before=before, allow_event=allow) return {"data": rows, "next": cursor} @router.get("/journal", tags=["status"], summary="The engine's journal events, newest first", responses=RESPONSES) @@ -840,8 +838,8 @@ async def get_journal( and its `payload`. Read only.""" require(principal, "view") events = await engine.journal.events_page(prefixes=[kind] if kind else (), before=before, limit=limit) - rows = [{"seq": e.seq, "ts": e.ts, "kind": e.kind, "key": e.key, "payload": e.payload} for e in events] - return {"data": rows, "next": rows[-1]["seq"] if len(rows) == limit else None} + rows = [{"seq": e.seq, "ts": e.ts, "kind": e.kind, "key": e.key, "payload": e.payload} for e in events if await may_view_event(principal, engine, e)] + return {"data": rows, "next": events[-1].seq if len(events) == limit else None} # -- risk and the kill switch -------------------------------------------- @@ -954,6 +952,8 @@ async def events(socket: WebSocket, key: str | None = Query(default=None), async for event in engine.bus.replay(since): if wanted and not subscription.wants(event): continue + if not await may_view_event(principal, engine, event): + continue await socket.send_text(json.dumps({ "seq": event.seq, "ts": event.ts, "kind": event.kind, "key": event.key, "payload": event.payload, }, default=str)) @@ -961,6 +961,8 @@ async def events(socket: WebSocket, key: str | None = Query(default=None), event = await subscription.queue.get() if wanted and not subscription.wants(event): continue + if not await may_view_event(principal, engine, event): + continue await socket.send_text(json.dumps({ "seq": event.seq, "ts": event.ts, "kind": event.kind, "key": event.key, "payload": event.payload, "dropped": subscription.dropped, diff --git a/synpath/trading/types.py b/synpath/trading/types.py index 1ccb488..0ff1a86 100644 --- a/synpath/trading/types.py +++ b/synpath/trading/types.py @@ -379,8 +379,22 @@ class Order(_Base): user_id: str | None = None id: str client_order_id: str | None = None - venue: str + venue: str | None account: Account | None = None + execution_accounts: dict[str, Account] = Field(default_factory=dict) + """Eligible execution accounts captured by a routed engine parent. + Its venue and singular account are null; actual children remain venue orders.""" + + @model_validator(mode="after") + def validate_execution_identity(self): + if self.execution_accounts: + if self.held_by != HeldBy.ENGINE or self.venue is not None or self.account is not None: + raise ValueError("cross-venue execution scope belongs to an engine parent without a singular venue/account") + if any(venue != account.venue for venue, account in self.execution_accounts.items()): + raise ValueError("execution account venue does not match its scope") + elif self.venue is None: + raise ValueError("an order without a venue needs its captured execution accounts") + return self market_id: str """Synpath id, `venue:native`.""" side: Side diff --git a/tests/test_journal_v3.py b/tests/test_journal_v3.py index e2e6468..6983843 100644 --- a/tests/test_journal_v3.py +++ b/tests/test_journal_v3.py @@ -64,7 +64,7 @@ class TestUpgradeFromTwo: async def test_a_fresh_file_is_the_current_version_with_the_new_shape(self, tmp_path: Path): path = str(tmp_path / "fresh.db") async with Journal(path) as journal: - assert await journal.schema_version() == SCHEMA_VERSION == 10 + assert await journal.schema_version() == SCHEMA_VERSION == 11 assert "parent_id" in columns(path, "orders") assert columns(path, "managed_orders")["account_key"] is False assert "bucket_id" in columns(path, "bucket_members") diff --git a/tests/test_owner_migrations.py b/tests/test_owner_migrations.py index b3ac2ff..d229a80 100644 --- a/tests/test_owner_migrations.py +++ b/tests/test_owner_migrations.py @@ -33,9 +33,9 @@ async def test_singleton_and_all_instruction_foreign_keys(config): async with create_journal(**config, user_id="owner-a") as store: async with store._db.execute("SELECT user_id FROM users") as cursor: assert [row["user_id"] for row in await cursor.fetchall()] == ["owner-a"] - assert await store.schema_version() == 10 + assert await store.schema_version() == 11 async with store._db.execute("SELECT version_num FROM alembic_version") as cursor: - assert (await cursor.fetchone())["version_num"] == "0005_drop_fair_values" + assert (await cursor.fetchone())["version_num"] == "0006_routed_execution_scope" for kind in ("market", "limit", "stop_market", "stop_limit"): req = request(kind, client_order_id="key-" + kind) await store.record_intent(req, client_order_id=req.client_order_id, venue="kalshi", account=Account(venue="kalshi")) @@ -95,7 +95,7 @@ async def test_baseline_adoption_preserves_or_refuses_without_relabeling(config, row = await store.typed_order("limit", "kalshi", "old") assert row["user_id"] == owner assert row["request"] == (body if has_instruction else None) - assert await store.schema_version() == 10 + assert await store.schema_version() == 11 async def test_buckets_stored_before_enabled_existed_load_with_every_member_on(config): @@ -106,7 +106,7 @@ async def test_buckets_stored_before_enabled_existed_load_with_every_member_on(c ("b", market_id, flip, position)) await connection.close() async with create_journal(**config, user_id="owner-a") as store: - assert await store.schema_version() == 10 + assert await store.schema_version() == 11 members = (await store.bucket("b"))["members"] assert [(m["market_id"], m["flip"], m["enabled"]) for m in members] == [ ("kalshi:KX-A", False, True), ("polymarket:1", True, True)] @@ -126,7 +126,7 @@ async def test_orders_from_before_0004_get_a_creation_time_and_page_newest_first (order_id, json.dumps(payload), created, updated)) await connection.close() async with create_journal(**config, user_id="owner-a") as store: - assert await store.schema_version() == 10 + assert await store.schema_version() == 11 rows = await store.orders_page(limit=2) assert [(order.id, key[0]) for order, key in rows] == [("undated", 9), ("dated", 5)] rest = await store.orders_page(limit=2, before=rows[-1][1]) @@ -169,7 +169,7 @@ async def pause(database, user_id): assert not store.is_open monkeypatch.setattr(migration, "upgrade", original) async with create_journal(**config, user_id="owner-a") as restarted: - assert await restarted.schema_version() == 10 + assert await restarted.schema_version() == 11 async with restarted._db.execute("SELECT user_id FROM users") as cursor: assert (await cursor.fetchone())["user_id"] == "owner-a" diff --git a/tests/test_postgres_storage.py b/tests/test_postgres_storage.py index 7c07958..1dbf3e6 100644 --- a/tests/test_postgres_storage.py +++ b/tests/test_postgres_storage.py @@ -373,7 +373,7 @@ async def fail(): assert await cursor.fetchone() is None monkeypatch.setattr(store, "_apply_schema", original) async with store: - assert await store.schema_version() == 10 + assert await store.schema_version() == 11 async def test_commit_failure_rolls_back_before_next_writer(journal, monkeypatch): diff --git a/tests/test_routed_audit_pagination.py b/tests/test_routed_audit_pagination.py new file mode 100644 index 0000000..5461d84 --- /dev/null +++ b/tests/test_routed_audit_pagination.py @@ -0,0 +1,195 @@ +"""Routed account isolation over sockets and indexed membership pagination.""" +from __future__ import annotations + +import base64 +import json + +import pytest +from fastapi.testclient import TestClient + +from synpath.engine.orders.routed import RoutedLimit +from synpath.engine.storage import create_journal +from synpath.server.store import ControlStore +from synpath.server.trading import create_trading_app +from synpath.trading.types import Account, Order +from test_owner_migrations import configuration, config # noqa: F401 +from test_postgres_storage import database # noqa: F401 +from test_routed_limit import build, request, ACCOUNT_K, ACCOUNT_P, K, P + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.parametrize("full_scope", [False, True], ids=["partial", "full"]) +async def test_socket_replay_and_live_filter_parents_but_keep_concrete_children(tmp_path, full_scope): + engine, _, _, _, bucket = await build(tmp_path) + store = await ControlStore(str(tmp_path / "socket-control.db")).open() + try: + parent = await engine.submit(request(bucket)) + owner, _ = await store.bootstrap("owner") + viewer = await store.create_user("viewer") + await store.acting_as(f"user:{owner.id}", "socket account grants") + await store.grant(viewer.id, ACCOUNT_K.key, "view") + if full_scope: + await store.grant(viewer.id, ACCOUNT_P.key, "view") + key = await store.issue_key(viewer.id) + children = [Order(id=f"probe-{venue}", venue=venue, account=account, + market_id=market, side="buy", type="limit", status="open", + time_in_force="gtc", quantity="1", price="0.4", parent_id=parent.id) + for venue, account, market in [("kalshi", ACCOUNT_K, K), ("polymarket", ACCOUNT_P, P)]] + + async def publish_phase(phase): + await engine.bus.publish("managed.security_probe", {**parent.model_dump(mode="json"), "probe": phase}, key=parent.id) + for child in children: + await engine.bus.publish("order.security_probe", {**child.model_dump(mode="json"), "probe": phase}, key=f"{child.venue}:{child.id}") + await engine.bus.publish("verification.barrier", {"probe": phase}) + + since = await engine.journal.last_seq() + await publish_phase("replay") + with TestClient(create_trading_app(engine, store)) as client: + kinds = "managed.security_probe,order.security_probe,verification.barrier" + with client.websocket_connect(f"/ws/events?key={key.secret}&since={since}&kinds={kinds}") as socket: + for phase in ("replay", "live"): + if phase == "live": + client.portal.call(publish_phase, phase) + received = [] + while True: + event = json.loads(socket.receive_text()) + assert event["payload"]["probe"] == phase + if event["kind"] == "verification.barrier": + break + received.append(event) + expected = {"probe-kalshi", "probe-polymarket", parent.id} if full_scope else {"probe-kalshi"} + assert {event["payload"]["id"] for event in received} == expected + assert len(received) == len(expected) + assert all(event["seq"] > since for event in received) + finally: + await store.close() + await engine.stop() + + +async def test_membership_filters_page_live_and_finished_parents_with_indexed_queries(config, tmp_path, monkeypatch): + engine, _, _, _, bucket = await build(tmp_path) + try: + accepted = await engine.submit(request(bucket)) + original = engine.managed_order_controller.get(accepted.id).snapshot() + async with create_journal(**config, user_id=engine.config.user_id) as store: + parents = [] + for index in range(8): + snapshot = {**original, "id": f"membership-{index}", "created_ms": 1000, + "state": "working" if index % 2 == 0 else "done"} + snapshot["request"] = {**original["request"], "client_order_id": f"membership-client-{index}"} + parent = RoutedLimit.from_snapshot(snapshot) + await store.save_managed(snapshot, order=parent.as_order()) + parents.append(parent.as_order()) + concrete = [] + for venue, account, market in [("kalshi", ACCOUNT_K, K), ("polymarket", ACCOUNT_P, P)]: + for terminal in (False, True): + order = Order(user_id=engine.config.user_id, id=f"concrete-{terminal}", venue=venue, + account=account, market_id=market, side="buy", type="limit", time_in_force="gtc", + status="closed" if terminal else "open", quantity="1", price="0.4", created_at=1000) + await store.upsert_order(order) + concrete.append(order) + outsider = Order(user_id=engine.config.user_id, id="outside", venue="other", account=Account(venue="other"), + market_id="other:market", side="buy", type="limit", time_in_force="gtc", status="open", quantity="1", created_at=1000) + await store.upsert_order(outsider) + captured = [] + executed_queries = [] + execute = store._db.execute + + def record_query(query, params=()): + executed_queries.append(query) + if query.startswith("SELECT payload") and "parent_execution_accounts" in query: + captured.append((query, params)) + return execute(query, params) + + monkeypatch.setattr(store._db, "execute", record_query) + for venue in ("kalshi", "polymarket"): + for open_only in (False, True): + for page_size in (1, 3, 20): + before_count = len(executed_queries) + first_page = await store.orders_page(venue=venue, open_only=open_only, limit=page_size) + queries = executed_queries[before_count:] + assert len(queries) == 1, queries + assert queries[0].startswith("SELECT payload") + assert "parent_execution_accounts" in queries[0] + assert len(first_page) == min(page_size, 5 if open_only else 10) + print(f"{config['backend']} venue={venue} open_only={open_only} limit={page_size}: " + f"{len(queries)} select, {len(first_page)} rows, no per-row scope queries") + rows, cursor = [], None + while True: + page = await store.orders_page(venue=venue, open_only=open_only, before=cursor, limit=3) + if not page: + break + rows.extend(page) + cursor = page[-1][1] + expected = [order for order in parents + concrete if (order.execution_accounts or order.venue == venue) + and (not open_only or not order.is_terminal)] + assert {(order.venue, order.id) for order, _ in rows} == {(order.venue, order.id) for order in expected} + assert len(rows) == len(expected) + # Venue/id ordering follows the database collation, which may + # differ from Python codepoint ordering (for example en_US). + # Page boundaries must reproduce that backend's complete order. + unpaged = await store.orders_page(venue=venue, open_only=open_only, limit=None) + assert [key for _, key in rows] == [key for _, key in unpaged] + assert sum(bool(order.execution_accounts) for order, _ in rows) == (4 if open_only else 8) + account = ACCOUNT_K if venue == "kalshi" else ACCOUNT_P + opened = await store.open_orders(account_key=account.key) + assert {(order.venue, order.id) for order in opened} == { + (order.venue, order.id) for order in parents + concrete + if not order.is_terminal and (order.execution_accounts or order.account == account)} + assert await store.open_orders(account_key="kalshi:unrelated") == [] + assert await store.orders_page(venue="unrelated", limit=3) == [] + monkeypatch.setattr(store._db, "execute", execute) + # Ask each backend to explain the actual production membership queries. + # Tiny PostgreSQL fixtures prefer sequential scans; disabling that option + # verifies index eligibility rather than claiming a production cost estimate. + if config["backend"] == "postgresql": + await store._db.execute("SET enable_seqscan=off") + for name, (query, params) in {"venue-page": next(item for item in captured if "created_ts" in item[0]), + "account-open": next(item for item in captured if "scope.account_key" in item[0])}.items(): + prefix = "EXPLAIN QUERY PLAN " if config["backend"] == "sqlite" else "EXPLAIN " + async with store._db.execute(prefix + query, params) as cursor: + plan = "\n".join(str(dict(row)) for row in await cursor.fetchall()) + assert "parent_execution" in plan + assert "INDEX" in plan.upper(), plan + assert "orders" in plan + assert any("parent_execution" in line and "INDEX" in line.upper() for line in plan.splitlines()), plan + print(f"{config['backend']} {name}: {plan}") + finally: + await engine.stop() + + +async def test_http_membership_pagination_rejects_old_cursor(tmp_path): + engine, _, _, _, bucket = await build(tmp_path) + store = await ControlStore(str(tmp_path / "cursor-control.db")).open() + try: + first = await engine.submit(request(bucket, client_order_id="page-first")) + await engine.cancel(first.id) + second = await engine.submit(request(bucket, client_order_id="page-second")) + _, key = await store.bootstrap("owner") + auth = {"Authorization": f"Bearer {key.secret}"} + with TestClient(create_trading_app(engine, store)) as client: + for venue in ("kalshi", "polymarket"): + seen, cursor = [], None + while True: + params = {"venue": venue, "scope": "all", "limit": 1} + if cursor: + params["cursor"] = cursor + response = client.get("/orders", params=params, headers=auth) + assert response.status_code == 200 + page = response.json() + seen.extend(order["id"] for order in page["data"]) + cursor = page["next_cursor"] + if not cursor: + break + assert set(seen) == {first.id, second.id} and len(seen) == 2 + old = base64.urlsafe_b64encode(json.dumps([1000, "kalshi", first.id]).encode()).decode().rstrip("=") + assert client.get("/orders", params={"scope": "all", "cursor": old}, headers=auth).status_code == 400 + finally: + await store.close() + await engine.stop() diff --git a/tests/test_routed_execution_scope.py b/tests/test_routed_execution_scope.py new file mode 100644 index 0000000..1aaf068 --- /dev/null +++ b/tests/test_routed_execution_scope.py @@ -0,0 +1,353 @@ +"""Cross-venue identity through real engine, HTTP, restart and both stores. + +All adapters are PaperVenue; these tests never use exchange credentials. +""" +from __future__ import annotations + +import json +from decimal import Decimal as D + +import pytest +from fastapi.testclient import TestClient + +from synpath.bucket import BucketMember +from synpath.engine import RiskConfig +from synpath.engine.storage import create_journal +from synpath.engine.storage_types import IntentState +from synpath.errors import BadRequest +from synpath.server.order_access import may_view_event +from synpath.server.store import ControlStore +from synpath.server.trading import create_trading_app +from synpath.trading.errors import RiskRejected, SubmissionPending +from synpath.trading.types import Account, Order, OrderRequest, OrderType, Side + +from test_routed_limit import build, book, request, K, P, ACCOUNT_K, ACCOUNT_P +from test_routed_recovery import resting_legs, new_engine +from test_owner_migrations import configuration, config # noqa: F401 +from test_postgres_storage import database # noqa: F401 + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.fixture +async def scope_engine(tmp_path): + engine, kalshi, poly, clock, bucket = await build(tmp_path) + book(kalshi, engine, K, asks=[(D("0.44"), D("100"))]) + book(poly, engine, P, asks=[(D("0.40"), D("100"))]) + yield engine, kalshi, poly, clock, bucket + await engine.stop() + + +async def test_reordering_members_preserves_owner_scope_and_nonfirst_allocation(scope_engine): + engine, kalshi, poly, _, bucket = scope_engine + first = await engine.submit(request(bucket, quantity=D("10"), client_order_id="original")) + reverse = bucket.model_copy(update={"members": list(reversed(bucket.members))}) + await engine.save_bucket(reverse) + second = await engine.submit(request(reverse, quantity=D("10"), client_order_id="reversed")) + assert first.venue is second.venue is None and first.account is second.account is None + assert first.execution_accounts == second.execution_accounts == {"kalshi": ACCOUNT_K, "polymarket": ACCOUNT_P} + assert first.user_id == second.user_id == engine.config.user_id + assert not kalshi.orders and [o.quantity for o in poly.orders.values()] == [D("10"), D("10")] + assert all(o.venue == "polymarket" and o.account == ACCOUNT_P for o in poly.orders.values()) + for venue in ("kalshi", "polymarket"): + rows = await engine.journal.orders_page(venue=venue) + assert sum(o.id == first.id for o, _ in rows) == 1 + assert (await engine.journal.order(None, first.id)).execution_accounts == first.execution_accounts + + +async def test_flipped_child_uses_member_account_and_native_side(tmp_path): + engine, kalshi, poly, _, bucket = await build(tmp_path, flip_p=True) + try: + book(poly, engine, P, bids=[(D("0.70"), D("100"))]) + book(kalshi, engine, K, asks=[(D("0.40"), D("100"))]) + parent = await engine.submit(request(bucket, quantity=D("10"), outcome="yes", action="buy")) + child = next(iter(poly.orders.values())) + assert parent.venue is None and not kalshi.orders + assert (child.account, child.side, child.price, child.outcome) == (ACCOUNT_P, Side.SELL, D("0.70"), "no") + finally: + await engine.stop() + + +@pytest.mark.parametrize("rule", ["capacity", "halt", "restricted"]) +async def test_nonfirst_destination_risk_refuses_before_dispatch(scope_engine, rule): + engine, kalshi, poly, _, bucket = scope_engine + if rule == "capacity": + # A 50/50 split fits but the agreed full-capacity policy refuses 100. + engine.order_risk_checks.configure(RiskConfig(max_position_contracts=D("60"), price_collar=None)) + elif rule == "halt": + engine.order_risk_checks.kill.engage(reason="test", scope="polymarket", policy="hold") + else: + engine.order_risk_checks.configure(RiskConfig(restricted=[P], price_collar=None)) + with pytest.raises(RiskRejected): + await engine.submit(request(bucket, quantity=D("100"))) + assert not kalshi.orders and not poly.orders + + +async def test_capture_is_used_once_across_authorization_and_concurrent_bucket_edit(scope_engine, monkeypatch): + engine, kalshi, poly, _, bucket = scope_engine + original_read = engine.journal.bucket + reads = [] + + async def read_then_edit(bucket_id): + row = await original_read(bucket_id) + reads.append(bucket_id) + await engine.save_bucket(bucket.model_copy(update={"members": [ + BucketMember(market_id=K), BucketMember(market_id="polymarket_us:new")]})) + return row + + monkeypatch.setattr(engine.journal, "bucket", read_then_edit) + authorized = [] + parent = await engine.submit(request(bucket), authorize_execution_accounts=lambda accounts: authorized.append(set(accounts))) + assert reads == [bucket.id] and authorized == [{"kalshi", "polymarket"}] + assert set(parent.execution_accounts) == {"kalshi", "polymarket"} + assert poly.orders and not kalshi.orders + + +async def test_same_client_id_retry_keeps_original_snapshot_after_bucket_edit(scope_engine): + engine, _, poly, _, bucket = scope_engine + instruction = request(bucket, client_order_id="retry") + first = await engine.submit(instruction) + await engine.save_bucket(bucket.model_copy(update={"members": [ + BucketMember(market_id=K), BucketMember(market_id=P, enabled=False)]})) + again = await engine.submit(instruction) + assert first.id == again.id and first.execution_accounts == again.execution_accounts + assert len(poly.orders) == 1 + with pytest.raises(BadRequest, match="different instruction"): + await engine.submit(instruction.model_copy(update={"quantity": D("1")})) + + +async def test_singular_parent_account_and_nullable_venue_order_are_rejected(scope_engine): + engine, _, _, _, bucket = scope_engine + with pytest.raises(BadRequest, match="singular account"): + await engine.submit(request(bucket, account=ACCOUNT_K)) + with pytest.raises(ValueError, match="captured execution accounts"): + Order(id="bad", venue=None, market_id=K, side="buy", type="limit", time_in_force="gtc", status="open", quantity="1") + + +async def test_legacy_client_id_without_verifiable_reservation_cannot_create_another_parent(scope_engine): + engine, _, poly, _, bucket = scope_engine + instruction = request(bucket, client_order_id="legacy-client") + parent = await engine.submit(instruction) + await engine.journal._db.execute("DELETE FROM submissions WHERE client_order_id=?", (instruction.client_order_id,)) + with pytest.raises(BadRequest, match="older bucket order"): + await engine.submit(instruction) + assert len(poly.orders) == 1 and (await engine._known(parent.id, None)).id == parent.id + + +@pytest.mark.parametrize("supplied_scope", [{"kalshi": ACCOUNT_K.model_dump(mode="json")}, {"kalshi": "malformed"}]) +async def test_caller_scope_params_never_authorize_the_full_reserved_request(scope_engine, tmp_path, supplied_scope): + engine, _, _, _, bucket = scope_engine + parent = await engine.submit(request(bucket, params={"execution_accounts": supplied_scope})) + store = await ControlStore(str(tmp_path / "audit-control.db")).open() + try: + owner, _ = await store.bootstrap("owner") + viewer = await store.create_user("partial") + await store.acting_as(f"user:{owner.id}", "test") + await store.grant(viewer.id, ACCOUNT_K.key, "view") + key = await store.issue_key(viewer.id) + principal = await store.principal(key.secret) + events = [event async for event in engine.journal.replay(0)] + reserved = next(event for event in events if event.kind == "submission.reserved") + assert set(reserved.payload["effective_request"]["params"]["execution_accounts"]) == {"kalshi", "polymarket"} + assert not await may_view_event(principal, engine, reserved) + with TestClient(create_trading_app(engine, store)) as client: + response = client.get("/journal", headers={"Authorization": f"Bearer {key.secret}"}) + assert response.status_code == 200 and parent.id not in response.text + assert "effective_request" not in response.text + finally: + await store.close() + + +async def test_changed_adapter_identity_blocks_bulk_reads_and_halt(scope_engine, monkeypatch): + engine, _, poly, _, bucket = scope_engine + engine.books.clear() + await engine.submit(request(bucket)) + # Keep the parent active but change only the actual adapter's account. + poly.account = Account(venue="polymarket", name="replacement") + calls = [] + + async def wrong_account(*args, **kwargs): + calls.append("wrong-account I/O") + raise AssertionError("must not contact replacement account") + + for name in ("fetch_open_orders", "fetch_my_trades", "fetch_positions", "fetch_balance", "cancel_all_orders"): + monkeypatch.setattr(poly, name, wrong_account) + with pytest.raises(SubmissionPending): + await engine.poll("polymarket") + from synpath.engine.reconcile import Reconciler + with pytest.raises(SubmissionPending): + await Reconciler(engine).run("polymarket") + result = await engine.halt("test", scope="polymarket") + assert "failed" in result["canceled"]["polymarket"] and calls == [] + + +async def test_parent_detail_reports_logs_and_legacy_events_use_saved_scope(scope_engine, tmp_path): + engine, _, _, _, bucket = scope_engine + parent = await engine.submit(request(bucket)) + store = await ControlStore(str(tmp_path / "control.db")).open() + try: + owner, owner_key = await store.bootstrap("owner") + partial = await store.create_user("partial") + await store.acting_as(f"user:{owner.id}", "test grants") + await store.grant(partial.id, ACCOUNT_K.key, "view") + await store.grant(partial.id, ACCOUNT_K.key, "trade") + key = await store.issue_key(partial.id, label="one venue") + partial_principal = await store.principal(key.secret) + auth = {"Authorization": f"Bearer {key.secret}"} + full = {"Authorization": f"Bearer {owner_key.secret}"} + with TestClient(create_trading_app(engine, store)) as client: + assert client.get(f"/orders/{parent.id}", headers=auth).status_code == 403 + assert client.delete(f"/orders/{parent.id}", headers=auth).status_code == 403 + assert client.patch(f"/orders/{parent.id}", headers=auth, json={"price": "0.43"}).status_code == 403 + assert client.get(f"/order-log?order_id={parent.id}", headers=auth).status_code == 403 + for path in ("/journal", "/order-log", "/orders?scope=all"): + assert parent.id not in json.dumps(client.get(path, headers=auth).json()) + assert client.get(f"/orders/{parent.id}", headers=full).json()["execution_accounts"]["polymarket"] + # The saved scope still governs history after editing the current definition. + await engine.save_bucket(bucket.model_copy(update={"members": [ + BucketMember(market_id=K), BucketMember(market_id=P, enabled=False)]})) + reports = client.get(f"/buckets/{bucket.id}/orders", headers=auth) + assert parent.id not in reports.text + assert client.get(f"/buckets/{bucket.id}/orders/{parent.id}", headers=auth).status_code == 403 + await store.grant(partial.id, ACCOUNT_P.key, "view") + await engine.save_bucket(bucket.model_copy(update={"members": [ + BucketMember(market_id=K), BucketMember(market_id="polymarket_us:later")]})) + # Access to the original history cannot depend on newly added accounts. + assert client.get(f"/buckets/{bucket.id}/orders/{parent.id}", headers=auth).status_code == 200 + assert parent.id in client.get(f"/buckets/{bucket.id}/orders", headers=auth).text + from synpath.engine.storage_types import JournalEvent + principal = partial_principal + legacy = JournalEvent(1, 1, "managed.working", parent.id, {"id": parent.id, "venue": "kalshi", "account": ACCOUNT_K.model_dump()}) + assert not await may_view_event(principal, engine, legacy) + finally: + await store.close() + + +async def test_changed_account_blocks_recovery_reads_cancels_and_false_lost(tmp_path, monkeypatch): + engine, kalshi, poly, clock, path, parent = await resting_legs(tmp_path) + managed = engine.managed_order_controller.get(parent.id) + child = managed.children[1] + child_order = await engine.journal.order(child.venue, child.order_id) + pending = OrderRequest(market_id=P, side=Side.BUY, quantity=D("1"), price=D("0.42"), + account=ACCOUNT_P, tags={"parent": parent.id}, client_order_id="uncertain") + await engine.journal.record_intent(pending, client_order_id="uncertain", venue="polymarket", account=ACCOUNT_P) + await engine.journal.mark_intent("uncertain", IntentState.SENDING) + await engine.journal.close() + calls = [] + + async def wrong_account(*args, **kwargs): + calls.append((args, kwargs)) + raise AssertionError("must not contact the replacement account") + + for name in ("fetch_order", "fetch_open_orders", "fetch_orders", "cancel_order"): + monkeypatch.setattr(poly, name, wrong_account) + restarted = new_engine(path, kalshi, poly, clock) + restarted.accounts["polymarket"] = Account(venue="polymarket", name="replacement") + await restarted.start() + try: + await restarted.managed_order_controller.on_timer() + with pytest.raises(SubmissionPending, match="differs"): + await restarted.cancel(child.order_id, venue=child.venue) + assert calls == [] + assert (await restarted.journal.intent("uncertain")).state == IntentState.SENDING + assert (await restarted.journal.order(child.venue, child.order_id)).status == child_order.status + assert restarted.managed_order_controller.get(parent.id).live + finally: + await restarted.stop() + + +@pytest.mark.parametrize("failure_boundary", ["membership", "projection", "event"]) +async def test_atomic_parent_snapshot_scope_projection_rollback_on_both_stores(config, tmp_path, monkeypatch, failure_boundary): + engine, _, _, _, bucket = await build(tmp_path) + try: + parent_order = await engine.submit(request(bucket, client_order_id="atomic")) + snapshot = engine.managed_order_controller.get(parent_order.id).snapshot() + async with create_journal(**config, user_id=engine.config.user_id) as store: + execute = store._db.execute + + def fail(query, params=()): + match = {"membership": "INSERT INTO parent_execution_accounts", "projection": "INSERT INTO orders(", + "event": "INSERT INTO events("}[failure_boundary] + if query.startswith(match): + raise RuntimeError("injected write failure") + return execute(query, params) + + monkeypatch.setattr(store._db, "execute", fail) + with pytest.raises(RuntimeError, match="injected"): + await store.save_managed(snapshot, order=parent_order) + monkeypatch.setattr(store._db, "execute", execute) + assert await store.managed_one(parent_order.id) is None + assert await store.order(None, parent_order.id) is None + assert await store.routed_account_keys("kalshi") == [] + assert await store.last_seq() == 0 + await store.save_managed(snapshot, order=parent_order) + assert (await store.order(None, parent_order.id)).execution_accounts == parent_order.execution_accounts + assert (await store.managed_one(parent_order.id))[1]["execution_accounts"] + finally: + await engine.stop() + + +@pytest.mark.parametrize("historical_evidence", ["complete", "missing", "explicit", "conflict"]) +async def test_legacy_nominal_parent_migration_preserves_or_refuses_accounts(config, tmp_path, historical_evidence): + engine, _, _, _, bucket = await build(tmp_path) + try: + parent_order = await engine.submit(request(bucket)) + snapshot = engine.managed_order_controller.get(parent_order.id).snapshot() + # Make an actual pre-PRE-6 snapshot and projection, keeping its original IDs. + snapshot.update(venue="kalshi", account=ACCOUNT_K.model_dump(mode="json")) + snapshot.pop("execution_accounts") + snapshot["request"]["account"] = ACCOUNT_K.model_dump(mode="json") + snapshot["request"]["params"].pop("execution_accounts") + legacy_projection = parent_order.model_dump(mode="json") + legacy_projection.update(venue="kalshi", account=ACCOUNT_K.model_dump(mode="json")) + legacy_projection.pop("execution_accounts") + async with create_journal(**config, user_id=engine.config.user_id) as store: + await store.save_managed(snapshot) + await store.upsert_order(Order.model_validate(legacy_projection)) + if historical_evidence in ("complete", "conflict"): + child = Order(id="historical-child", venue="polymarket", account=ACCOUNT_P, market_id=P, + side="buy", type="limit", time_in_force="gtc", status="open", quantity="1", + tags={"parent": parent_order.id}, user_id=engine.config.user_id) + await store.upsert_order(child) + if historical_evidence in ("explicit", "conflict"): + account = ACCOUNT_P if historical_evidence == "explicit" else Account(venue="polymarket", name="conflict") + await store._db.execute("INSERT INTO meta(key,value) VALUES('legacy_execution_accounts',?)", + (json.dumps({parent_order.id: {"polymarket": account.model_dump(mode="json")}}),)) + await store._db.execute("DROP TABLE parent_execution_accounts") + await store._db.execute("UPDATE alembic_version SET version_num='0005_drop_fair_values'") + await store._db.execute("UPDATE meta SET value='10' WHERE key='schema_version'") + reopened = create_journal(**config, user_id=engine.config.user_id) + if historical_evidence in ("missing", "conflict"): + with pytest.raises(ValueError, match="historical execution accounts|conflicting historical"): + await reopened.open() + assert not reopened.is_open + # Retry using a reviewed explicit mapping for the missing venue. + if historical_evidence == "missing": + from synpath.engine.database import connect_sqlite, connect_postgres + connection = await (connect_sqlite(config["path"]) if config["backend"] == "sqlite" + else connect_postgres(config["path"])) + if config["backend"] == "postgresql": + await connection.execute(f'SET search_path TO "{config["schema"]}", pg_catalog') + async with connection.execute("SELECT value FROM meta WHERE key='schema_version'") as cursor: + assert (await cursor.fetchone())["value"] == "10" + await connection.execute("INSERT INTO meta(key,value) VALUES('legacy_execution_accounts',?)", + (json.dumps({parent_order.id: {"polymarket": ACCOUNT_P.model_dump(mode="json")}}),)) + await connection.close() + else: + return + async with reopened: + migrated = await reopened.order(None, parent_order.id) + assert migrated.id == parent_order.id and migrated.venue is None and migrated.account is None + assert migrated.execution_accounts == {"kalshi": ACCOUNT_K, "polymarket": ACCOUNT_P} + assert (await reopened.managed_one(parent_order.id))[1]["legacy_execution_identity"]["venue"] == "kalshi" + assert await reopened.order("kalshi", parent_order.id) is None + assert sum(order.id == parent_order.id for order, _ in await reopened.orders_page()) == 1 + async with reopened: + assert (await reopened.order(None, parent_order.id)).execution_accounts == migrated.execution_accounts + finally: + await engine.stop() diff --git a/tests/test_routed_scope_transactions.py b/tests/test_routed_scope_transactions.py new file mode 100644 index 0000000..358eef0 --- /dev/null +++ b/tests/test_routed_scope_transactions.py @@ -0,0 +1,131 @@ +"""Post-write rollback and executable legacy routed-parent upgrade contracts.""" +import asyncio +import copy +from decimal import Decimal as D + +import pytest + +from synpath.engine import Engine, EngineConfig, RiskConfig +from synpath.engine.storage import create_journal +from synpath.trading.types import Order, OrderRequest, Side +from test_owner_migrations import configuration, config # noqa: F401 +from test_postgres_storage import database # noqa: F401 +from test_routed_limit import build, request, ACCOUNT_K, ACCOUNT_P, K +from test_routed_recovery import resting_legs + +pytestmark = pytest.mark.anyio + + +@pytest.fixture +def anyio_backend(): + return "asyncio" + + +@pytest.mark.parametrize("operation", ["save", "plan"]) +@pytest.mark.parametrize("interruption", ["failure", "cancellation"]) +async def test_parent_transaction_rolls_back_after_all_component_writes(config, tmp_path, monkeypatch, operation, interruption): + source, _, _, _, bucket = await build(tmp_path) + try: + original = request(bucket, client_order_id="post-write-atomic", user_id=source.config.user_id) + parent = await source.submit(original) + snapshot = source.managed_order_controller.get(parent.id).snapshot() + effective = OrderRequest.stored(snapshot["request"]) + child = OrderRequest(market_id=K, side="buy", quantity="1", price="0.40", account=ACCOUNT_K, + user_id=parent.user_id, client_order_id="planned-child", tags={"parent": parent.id}) + store = create_journal(**config, user_id=parent.user_id) + async with store: + await store.record_submission(original, effective, venue="__managed__", account=None) + baseline_seq = await store.last_seq() + entered = asyncio.Event() + helper_name = "_write_order" if operation == "save" else "_write_intent" + helper = getattr(store, helper_name) + + async def interrupt_after_write(*args, **kwargs): + await helper(*args, **kwargs) + entered.set() + if interruption == "failure": + raise RuntimeError("failure after all component writes") + await asyncio.Event().wait() + + monkeypatch.setattr(store, helper_name, interrupt_after_write) + action = (store.save_managed(snapshot, order=parent) if operation == "save" else + store.plan_managed_child(snapshot, child, venue="kalshi", account=ACCOUNT_K, parent_order=parent)) + task = asyncio.create_task(action) + await asyncio.wait_for(entered.wait(), 5) + if interruption == "cancellation": + task.cancel() + with pytest.raises(asyncio.CancelledError if interruption == "cancellation" else RuntimeError): + await task + monkeypatch.setattr(store, helper_name, helper) + assert await store.managed_one(parent.id) is None + assert await store.order(None, parent.id) is None + assert await store.routed_account_keys("kalshi") == [] + assert await store.intent("planned-child") is None + assert await store.last_seq() == baseline_seq + reservation = await store.submission(original.client_order_id) + assert reservation["result_id"] is None and reservation["state"] == "pending" + async with store: + assert await store.managed_one(parent.id) is None + assert await store.order(None, parent.id) is None + assert (await store.submission(original.client_order_id))["result_id"] is None + await store.save_managed(snapshot, order=parent) + assert (await store.submission(original.client_order_id))["result_id"] == parent.id + assert (await store.order(None, parent.id)).execution_accounts == parent.execution_accounts + finally: + await source.stop() + + +async def test_migrated_working_parent_restores_and_cancels_original_children(config, tmp_path): + source, kalshi, poly, clock, _, original = await resting_legs(tmp_path) + try: + kalshi.on_trade(K, price=D("0.41"), quantity=D("10"), taker_side=Side.SELL) + await kalshi.deliver() + managed = source.managed_order_controller.get(original.id) + assert managed.filled == D("10") and managed.state == "working" + snapshot = copy.deepcopy(managed.snapshot()) + projection = managed.as_order().model_dump(mode="json") + children = await source.journal.orders_for_parent(original.id) + snapshot.update(venue="kalshi", account=ACCOUNT_K.model_dump(mode="json")) + snapshot.pop("execution_accounts") + snapshot["request"]["account"] = ACCOUNT_K.model_dump(mode="json") + snapshot["request"]["params"].pop("execution_accounts") + projection.update(venue="kalshi", account=ACCOUNT_K.model_dump(mode="json")) + projection.pop("execution_accounts") + async with create_journal(**config, user_id=original.user_id) as store: + await store.save_managed(snapshot) + await store.upsert_order(Order.model_validate(projection)) + for child in children: + await store.upsert_order(child) + await store._db.execute("DROP TABLE parent_execution_accounts") + await store._db.execute("UPDATE alembic_version SET version_num='0005_drop_fair_values'") + await store._db.execute("UPDATE meta SET value='10' WHERE key='schema_version'") + finally: + await source.stop() + + journal = create_journal(**config, user_id=original.user_id) + restored = Engine({"kalshi": kalshi, "polymarket": poly}, + EngineConfig(user_id=original.user_id, require_lease=True), journal=journal, + accounts={"kalshi": ACCOUNT_K, "polymarket": ACCOUNT_P}, clock=clock, + risk=RiskConfig(price_collar=None, closing_soon_s=None)) + for paper in (kalshi, poly): + paper.listeners.clear() + paper.order_listeners.clear() + paper.subscribe(restored.on_fill) + paper.subscribe_orders(restored.on_order) + await restored.start() + try: + parent = restored.managed_order_controller.get(original.id) + assert parent is not None and parent.live + assert parent.execution_accounts == {"kalshi": ACCOUNT_K, "polymarket": ACCOUNT_P} + assert parent.filled == D("10") + assert {(c.venue, c.order_id) for c in parent.children} == {(c.venue, c.id) for c in children} + result = await restored.cancel(original.id) + assert result.status.value == "canceled" and result.filled == D("10") + assert not [order for paper in (kalshi, poly) for order in paper.orders.values() if not order.is_terminal] + assert len(kalshi.orders) == len(poly.orders) == 1 + for child in await journal.orders_for_parent(original.id): + assert child.account == {"kalshi": ACCOUNT_K, "polymarket": ACCOUNT_P}[child.venue] + assert (await journal.order(None, original.id)).filled == D("10") + assert await journal.order("kalshi", original.id) is None + finally: + await restored.stop() diff --git a/tests/test_server_trading.py b/tests/test_server_trading.py index 662b6ad..a3a857b 100644 --- a/tests/test_server_trading.py +++ b/tests/test_server_trading.py @@ -565,7 +565,7 @@ async def test_accounts_lists_a_configured_venue_without_a_named_account(self, t try: with TestClient(create_trading_app(engine, store)) as client: rows = client.get("/accounts", headers=auth(key.secret)).json() - assert [r["account"] for r in rows] == ["kalshi:default"] + assert [r["account"] for r in rows] == ["kalshi:paper"] # the adapter's actual execution identity finally: await store.close() await engine.stop()