fix(ledger): carry the owning rules.id on a position so exits are attributed (#803)

keel/agent.py

@@ -387,6 +387,12 @@ def _open_tranche(
         entry_fee=order["fee"] or Decimal("0"),
         entry_fill=entry_fill,
         initial_stop=initial_stop,
+        # Inherited from the ENTRY order rather than passed down separately (#803). That order
+        # was written with `rule_id=signal.rule_id`, so it already carries the identity of the
+        # row that fired -- and taking it from here means every caller of `_open_tranche` gets
+        # the attribution without a new argument to forget. `None` stays None: an order that
+        # recorded no id cannot be given one now.
+        rule_id=order.get("rule_id"),
     )
     if result.bracket_order_id is not None:
         repo.set_position_bracket(position_id, result.bracket_order_id)

keel/data/db.py

@@ -20,7 +20,7 @@
 from pathlib import Path
 from typing import Any
 
-SCHEMA_VERSION = 20
+SCHEMA_VERSION = 21
 
 # Creation order matters for readability (and for backends that validate FK targets eagerly);
 # SQLite itself only checks FK targets at DML time, but we still declare referenced tables first.
@@ -135,6 +135,26 @@
         entry_fill        TEXT    NOT NULL,
         entry_fee         TEXT    NOT NULL,
         initial_stop      TEXT,
+        -- The `rules.id` that opened this tranche (#803). `rule_name` above is the rule's KIND
+        -- ("dca", "turtle_breakout"), which is what the engine reconstructs a Rule from -- it
+        -- does not identify the ROW, so nothing could thread a rules.id onto the protective
+        -- SELL that later exits this position. Every bracket and exit order was therefore
+        -- written with `orders.rule_id` NULL and rendered "unattributed", leaving per-rule
+        -- accounting with entries attributed and exits anonymous -- the wrong half to lose.
+        --
+        -- NULL means unknown, never "no rule": tranches opened before v21 have no recorded id
+        -- and there is nothing to join them back through (`positions` references its BRACKET
+        -- order, never its ENTRY order), so they are left NULL rather than guessed at. Same
+        -- rule v12 set for `initial_stop`.
+        --
+        -- NO FOREIGN KEY, unlike every other `rule_id` in this schema (`orders` and both
+        -- attestation tables declare one). The asymmetry is about PARITY, not about caring
+        -- less: SQLite cannot carry a constraint through `ALTER TABLE ... ADD COLUMN`, so a
+        -- database reaching v21 by MIGRATION physically cannot have one. Declaring it here
+        -- only would mean a fresh database enforces a rule a migrated one does not -- the two
+        -- diverge silently, and the difference surfaces as an IntegrityError on one deployment
+        -- and not another. Readers must treat this id as a hint that may name a deleted row.
+        rule_id           INTEGER,
         -- Partial-exit accumulators (#502). `qty` is the quantity STILL HELD, and it is now
         -- mutable: `scale_out` sells a fraction of a tranche and leaves the rest running, so
         -- the legs of one trade land at different prices and different times. These three
@@ -1024,6 +1044,39 @@ def _migrate_v20_provenance_and_attest_windows(conn: sqlite3.Connection) -> None
         conn.execute("ALTER TABLE instrument_attestations ADD COLUMN attest_due_ts INTEGER")
 
 
+def _migrate_v21_positions_rule_id(conn: sqlite3.Connection) -> None:
+    """v21 adds `positions.rule_id` -- the `rules.id` that opened the tranche (#803).
+
+    `positions.rule_name` is a KIND, not an identity. It is what `agent._build_rule`
+    reconstructs a Rule from, and it was enough for everything the ledger did until an EXIT
+    needed attributing: `executor.place_bracket` and the scale-out path build their
+    `OrderIntent` from a POSITION rather than from a signal, so they had no `rules.id` to put on
+    `orders.rule_id`. Every protective SELL was written unattributed, and
+    `payload._order_payload` rendered it as such.
+
+    That is not a cosmetic gap. Per-rule accounting reading `orders.rule_id` saw entries
+    attributed and exits anonymous, which for a trend rule drops exactly the leg the outcome
+    lands on.
+
+    Idempotent by the usual `PRAGMA table_info` guard: a database stamped at v20 got `positions`
+    from v4's DDL, and `CREATE TABLE IF NOT EXISTS` never adds a column to an existing table.
+
+    **NO BACKFILL, deliberately** -- the same call v12 made for `initial_stop`. A tranche records
+    its BRACKET order (`bracket_order_id`), never its ENTRY order, so there is no join back to
+    the order that carries the id. Matching on `(product_id, rule_name)` would re-attribute by
+    guess and would be wrong wherever a rule row was replaced. NULL means "nobody recorded it",
+    and readers must show it as unknown rather than invent an owner.
+
+    **No foreign key**, unlike `orders.rule_id`. `ALTER TABLE ... ADD COLUMN` cannot carry a
+    constraint in SQLite, so a database reaching v21 by migration could not have one; declaring
+    it on the fresh DDL alone would leave fresh and migrated deployments enforcing different
+    rules. See the column comment in the `positions` DDL for the full reasoning.
+    """
+    columns = {row["name"] for row in conn.execute("PRAGMA table_info(positions)")}
+    if "rule_id" not in columns:
+        conn.execute("ALTER TABLE positions ADD COLUMN rule_id INTEGER")
+
+
 _MIGRATIONS: dict[int, Callable[[sqlite3.Connection], None]] = {
     2: _migrate_v2_broker_subscriptions,
     3: _migrate_v3_trade_outcomes,
@@ -1044,6 +1097,7 @@ def _migrate_v20_provenance_and_attest_windows(conn: sqlite3.Connection) -> None
     18: _migrate_v18_venue_cash_postures,
     19: _migrate_v19_equity_points,
     20: _migrate_v20_provenance_and_attest_windows,
+    21: _migrate_v21_positions_rule_id,
 }
 
 

keel/data/repository.py

@@ -1253,6 +1253,7 @@ def open_position(
         entry_fee: Decimal,
         initial_stop: Decimal | None = None,
         bracket_order_id: int | None = None,
+        rule_id: int | None = None,
     ) -> int:
         """Record a newly opened tranche and return its id.
 
@@ -1262,13 +1263,20 @@ def open_position(
         before v12 predates the column. Readers must disable the break-even arm for such a
         tranche rather than substitute the current stop, which is a different policy (see
         `db._migrate_v12_positions_initial_stop`).
+
+        `rule_id` is the `rules.id` that opened the tranche (#803), distinct from `rule_name`,
+        which is the rule's KIND. Only the id identifies the ROW, and it is what lets the
+        protective SELL that later exits this position be attributed: both exit paths build
+        their `OrderIntent` from a position, not from a signal, so without this they wrote
+        `orders.rule_id` NULL and rendered "unattributed". `None` means unknown -- pre-v21
+        tranches, and any entry whose order carried no id -- never "no owning rule".
         """
         cursor = self._conn.execute(
             """
             INSERT INTO positions
                 (product_id, rule_name, opened_at, qty, entry_fill, entry_fee,
-                 initial_stop, bracket_order_id, status)
-            VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'open')
+                 initial_stop, bracket_order_id, rule_id, status)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'open')
             """,
             (
                 product_id,
@@ -1279,6 +1287,7 @@ def open_position(
                 _dec_to_text(entry_fee),
                 None if initial_stop is None else _dec_to_text(initial_stop),
                 bracket_order_id,
+                rule_id,
             ),
         )
         self._conn.commit()

keel/execution/executor.py

@@ -242,6 +242,7 @@ def execute(
             target=signal.setup.target,
             rule_name=signal.rule_name,
             now_ts=now_ts,
+            rule_id=signal.rule_id,  # #803 -- the same id the ENTRY order was written with
         )
         # Surfaced rather than discarded so `run_once` can point the tranche at its bracket.
         # See `ExecutionResult.bracket_order_id`.
@@ -2146,6 +2147,7 @@ def place_bracket(
     target: Decimal,
     rule_name: str,
     now_ts: int,
+    rule_id: int | None = None,
 ) -> int | None:
     """Place the exchange-side exit bracket for an open long position, or `None` if vetoed.
 
@@ -2177,6 +2179,11 @@ def place_bracket(
         notional=sizing.spend(qty, stop),
         is_dca=False,
         rule_kind=rule_name,
+        # #803: `rule_kind` is the KIND, and it is not an identity. Without the id the row this
+        # order belongs to is unrecoverable from the orders ledger, so every protective SELL
+        # read "unattributed" while its entry read the rule's name. Callers that hold a position
+        # pass `position["rule_id"]`; `None` stays None rather than being guessed from the kind.
+        rule_id=rule_id,
         available_base=held,
     )
     # Built BEFORE the call, and inside a try, deliberately. As an inline argument to
@@ -2268,6 +2275,7 @@ def scale_out(
     exit_price: Decimal,
     rule_name: str,
     now_ts: int,
+    rule_id: int | None = None,
 ) -> ExecutionResult:
     """Sell `qty` of an open position and RESIZE its protective bracket down to the remainder.
 
@@ -2411,6 +2419,7 @@ def scale_out(
         notional=sizing.spend(qty, exit_price),
         is_dca=False,
         rule_kind=rule_name,
+        rule_id=rule_id,  # #803, as in `place_bracket` above
         available_base=venue_held,
     )
     result = _run_order(intent, broker, repo, config, "autonomous", None, now_ts)
@@ -2453,6 +2462,7 @@ def scale_out(
         target=target,
         rule_name=rule_name,
         now_ts=now_ts,
+        rule_id=rule_id,  # #803 -- the resized bracket belongs to the same rule as the scale-out
     )
     if bracket_order_id is None:
         log_event(
@@ -2583,6 +2593,10 @@ def _roll_stop(
         )
         return None
 
+    # Read BEFORE the cancel-and-replace below, while the tranche still names the OLD bracket:
+    # the success path repoints it to the replacement, after which this id resolves to nothing.
+    rolled_position = repo.get_position_for_bracket(old_stop_order_id)
+
     target = repo.get_state(f"open_target:{product_id}")
     if target is None:
         log_event(
@@ -2682,6 +2696,14 @@ def _roll_stop(
         notional=sizing.spend(qty, new_stop),
         is_dca=False,
         rule_kind=rule_name,
+        # #803: resolved HERE rather than threaded through `_roll_stop`'s three public wrappers,
+        # because this function already holds the one thing that identifies the owner --
+        # `old_stop_order_id`, the bracket the tranche currently names. Reading the ledger is
+        # also the more honest source than a `rule_id` passed down a call chain: it is the same
+        # lookup the success path below already does to repoint the tranche. `None` (a tranche
+        # predating v21, or a bracket no tranche names) stays None, exactly as `rule_kind` alone
+        # behaved before -- a roll must never fail over missing attribution.
+        rule_id=(rolled_position or {}).get("rule_id"),
         available_base=held,
     )
     # Built BEFORE the call and inside a try, for the same reason `place_bracket` is -- and more

keel/execution/reconcile.py

@@ -290,6 +290,7 @@ def reconcile_unbracketed_positions(
                 target=intent["target"],
                 rule_name=position.get("rule_name") or "rebracket",
                 now_ts=now_ts,
+                rule_id=position.get("rule_id"),  # #803
             )
         except Exception:
             log_exception(
@@ -549,6 +550,7 @@ def _rebracket_or_escalate(
             target=target,
             rule_name=position.get("rule_name") or "rebracket",
             now_ts=now_ts,
+            rule_id=position.get("rule_id"),  # #803
         )
     except Exception:
         log_exception(

tests/data/test_db.py

@@ -103,11 +103,11 @@ def test_agent_state_table_has_key_primary_key():
     assert pk_columns == {"key"}
 
 
-def test_schema_version_is_20():
+def test_schema_version_is_21():
     """Deliberate tripwire: bump this literal consciously on every schema change."""
     from keel.data.db import SCHEMA_VERSION
 
-    assert SCHEMA_VERSION == 20
+    assert SCHEMA_VERSION == 21
 
 
 def test_a_v6_database_migrates_up_and_gains_the_profile_table(tmp_path):

tests/data/test_migrations.py

@@ -49,7 +49,7 @@ def test_fresh_database_is_stamped_at_the_current_version() -> None:
     conn = db.connect(":memory:")
     db.migrate(conn)
     version = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert version == db.SCHEMA_VERSION == 20
+    assert version == db.SCHEMA_VERSION == 21
 
 
 def test_fresh_database_gets_no_subscription_row() -> None:
@@ -624,7 +624,7 @@ def test_v14_migration_bumps_the_stored_version() -> None:
     conn = _v12_database()
     db.migrate(conn)
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_v14_migration_step_is_not_blocked_by_another_venues_existing_row() -> None:
@@ -785,7 +785,7 @@ def test_v15_migration_bumps_the_stored_version() -> None:
     conn = _v12_database()
     db.migrate(conn)
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_v15_the_12_to_15_chain_creates_the_table_with_the_column_already_present() -> None:
@@ -887,7 +887,7 @@ def test_an_existing_orders_table_gains_the_submit_book_by_ALTER() -> None:
     assert row["submit_best_bid"] is None
     assert row["submit_best_ask"] is None
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_v16_is_idempotent_per_column() -> None:
@@ -1047,7 +1047,7 @@ def test_migration_to_v20_adds_the_columns_and_the_new_tables() -> None:
     assert "idx_cycle_balances_mode_currency_ts" in index_names
 
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_v19_database_gains_v20_columns_as_NULL_no_backfill() -> None:
@@ -1094,7 +1094,7 @@ def test_v19_database_gains_v20_columns_as_NULL_no_backfill() -> None:
     assert instrument_row["attest_due_ts"] is None
 
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_v20_is_idempotent_per_column() -> None:
@@ -1165,7 +1165,7 @@ def test_v20_on_a_pre_v11_chain_does_not_duplicate_columns() -> None:
     assert {"cycle_balances", "audit_events"} <= table_names
 
     stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
-    assert stamped == db.SCHEMA_VERSION == 20
+    assert stamped == db.SCHEMA_VERSION == 21
 
 
 def test_cycle_balances_accepts_null_and_round_trips_a_decimal_string() -> None:
@@ -1249,3 +1249,80 @@ def test_the_recorded_balance_columns_are_text_and_nullable() -> None:
 
     for name in ("available", "total"):
         assert columns[name] == ("TEXT", 0), f"cycle_balances.{name} is {columns[name]}"
+
+
+# -- v21: positions.rule_id (#803) -------------------------------------------------------------
+
+
+def _v20_positions_table(conn: sqlite3.Connection) -> None:
+    """`positions` exactly as v20 shipped it -- every column through `status`, and no `rule_id`."""
+    conn.execute(
+        """
+        CREATE TABLE positions (
+            id                INTEGER PRIMARY KEY AUTOINCREMENT,
+            product_id        TEXT    NOT NULL,
+            rule_name         TEXT    NOT NULL,
+            opened_at         INTEGER NOT NULL,
+            closed_at         INTEGER,
+            qty               TEXT    NOT NULL,
+            entry_fill        TEXT    NOT NULL,
+            entry_fee         TEXT    NOT NULL,
+            initial_stop      TEXT,
+            realized_qty      TEXT,
+            realized_proceeds TEXT,
+            realized_fees     TEXT,
+            bracket_order_id  INTEGER,
+            status            TEXT    NOT NULL DEFAULT 'open'
+        )
+        """
+    )
+
+
+def test_migration_to_v21_adds_positions_rule_id() -> None:
+    conn = db.connect(":memory:")
+    db.migrate(conn)
+
+    columns = {row["name"] for row in conn.execute("PRAGMA table_info(positions)")}
+
+    assert "rule_id" in columns
+    assert "rule_name" in columns, "the KIND must survive alongside the id, not be replaced by it"
+
+
+def test_a_v20_database_gains_rule_id_as_NULL_no_backfill() -> None:
+    """NO BACKFILL, the call v12 made for `initial_stop`.
+
+    A tranche records its BRACKET order, never its ENTRY order, so there is no join back to the
+    row carrying the id. Matching on `(product_id, rule_name)` would re-attribute by guess and be
+    wrong wherever a rule row was replaced. NULL means "nobody recorded it".
+    """
+    conn = db.connect(":memory:")
+    conn.execute("CREATE TABLE IF NOT EXISTS schema_version (version INTEGER NOT NULL)")
+    conn.execute("INSERT INTO schema_version (version) VALUES (20)")
+    _v20_positions_table(conn)
+    conn.execute(
+        """
+        INSERT INTO positions (product_id, rule_name, opened_at, qty, entry_fill, entry_fee)
+        VALUES ('PAXG-USD', 'turtle_breakout', 1, '0.0132', '4673.23', '0.73')
+        """
+    )
+    conn.commit()
+
+    db.migrate(conn)
+
+    row = conn.execute("SELECT rule_name, rule_id FROM positions").fetchone()
+    assert row["rule_name"] == "turtle_breakout"
+    assert row["rule_id"] is None, "a pre-v21 tranche must read unknown, not a guessed owner"
+    stamped = conn.execute("SELECT version FROM schema_version").fetchone()["version"]
+    assert stamped == db.SCHEMA_VERSION == 21
+
+
+def test_v21_is_idempotent_on_a_database_that_already_has_the_column() -> None:
+    """`CREATE TABLE IF NOT EXISTS` never adds a column, so the ALTER is guarded by
+    `PRAGMA table_info` -- running it twice must not raise `duplicate column name`."""
+    conn = db.connect(":memory:")
+    db.migrate(conn)
+
+    db._migrate_v21_positions_rule_id(conn)
+
+    columns = [row["name"] for row in conn.execute("PRAGMA table_info(positions)")]
+    assert columns.count("rule_id") == 1