fix(retrieval): widen outbox aggregate IDs

packages/infinity_context_adapters/infinity_context_adapters/postgres/migrations/0060_memory_outbox_aggregate_id_width.sql

@@ -0,0 +1,5 @@
+SET LOCAL lock_timeout = '5s';
+SET LOCAL statement_timeout = '30s';
+
+ALTER TABLE public.memory_outbox
+  ALTER COLUMN aggregate_id TYPE VARCHAR(120);

packages/infinity_context_adapters/infinity_context_adapters/postgres/outbox_models.py

@@ -34,7 +34,7 @@ class MemoryOutboxRow(Base):
     message_key: Mapped[str | None] = mapped_column(String(160), nullable=True)
     event_type: Mapped[str] = mapped_column(String(120), nullable=False)
     aggregate_type: Mapped[str] = mapped_column(String(80), nullable=False)
-    aggregate_id: Mapped[str] = mapped_column(String(80), nullable=False)
+    aggregate_id: Mapped[str] = mapped_column(String(120), nullable=False)
     aggregate_version: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
     workload_class: Mapped[str] = mapped_column(String(80), nullable=False, default="projection")
     fairness_key: Mapped[str | None] = mapped_column(String(160), nullable=True)

tests/e2e/managed_cleanup_v3_full_postgres_support.py

@@ -473,7 +473,7 @@ async def create_full_postgres_harness(database_url: str, work_dir: Path) -> Ful
         engine = build_async_engine(database.app_url)
         try:
             result = await upgrade_schema(engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
         finally:
             await engine.dispose()
         context, authority, pages, operations = build_strict_v4_material()

tests/e2e/test_locator_parent_lifecycle_postgres.py

@@ -89,7 +89,10 @@ async def _assert_upgrade_and_fresh(database_url: str) -> None:
                 await connection.close()
 
             result = await upgrade_schema(engine)
-            assert result.applied == ("0059_locator_parent_lifecycle",)
+            assert result.applied == (
+                "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
+            )
             await _assert_staged_repair(engine)
             await _assert_benchmark_fence_persistence(engine, asyncpg)
             await build_locator_retrieval_indexes(engine, statement_timeout_ms=30_000)
@@ -106,7 +109,7 @@ async def _assert_upgrade_and_fresh(database_url: str) -> None:
         engine = build_async_engine(database.app_url)
         try:
             fresh = await upgrade_schema(engine)
-            assert fresh.current == "0059_locator_parent_lifecycle"
+            assert fresh.current == "0060_memory_outbox_aggregate_id_width"
             assert fresh.applied[0] == "0001_core_facts"
             await build_locator_retrieval_indexes(engine, statement_timeout_ms=30_000)
             await _assert_catalog(engine)

tests/e2e/test_locator_parent_retraction_postgres.py

@@ -41,7 +41,10 @@ async def _assert_parent_retraction(database_url: str) -> None:
         engine = build_async_engine(database.app_url)
         try:
             upgraded = await upgrade_schema(engine)
-            assert upgraded.applied == ("0059_locator_parent_lifecycle",)
+            assert upgraded.applied == (
+                "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
+            )
             await _assert_coordinate_edit_egress(engine)
             await _assert_classification_tightening(engine, asyncpg)
             await _assert_owner_retraction(engine, asyncpg)

tests/e2e/test_locator_profile_0046_populated_upgrade_postgres.py

@@ -121,6 +121,7 @@ async def _assert_populated_upgrade(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             registry = PostgresRetrievalProfileRegistry(build_session_factory(engine))
             blocker = await engine.connect()

tests/e2e/test_locator_profile_outbox_transaction_coalescing_postgres.py

@@ -43,6 +43,7 @@ async def _assert_upgrade_and_coalescing(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             assert (await upgrade_schema(engine)).applied == ()
         finally:

tests/e2e/test_locator_retrieval_transit_versions_postgres.py

@@ -88,6 +88,7 @@ async def _scenario(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             await _assert_transit_column_types(engine)
 
@@ -344,6 +345,7 @@ async def _upgrade_repair_scenario(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             async with engine.connect() as connection:
                 tombstone = (

tests/e2e/test_locator_retrieval_upgrade_postgres.py

@@ -76,11 +76,12 @@ async def _assert_cutover_upgrade(database_url: str, starting_migration: str) ->
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             installed_count = 1 if starting_migration == "0039_" else 2
             assert upgraded.applied == expected[installed_count:]
-            assert upgraded.current == "0059_locator_parent_lifecycle"
-            assert upgraded.applied[-1] == "0059_locator_parent_lifecycle"
+            assert upgraded.current == "0060_memory_outbox_aggregate_id_width"
+            assert upgraded.applied[-1] == "0060_memory_outbox_aggregate_id_width"
             assert len(await build_locator_retrieval_indexes(engine)) == 3
             assert len(await build_locator_retrieval_indexes(engine)) == 3
             async with engine.begin() as connection:

tests/e2e/test_memory_outbox_aggregate_id_width_postgres.py

@@ -0,0 +1,73 @@
+from __future__ import annotations
+
+import asyncio
+import os
+
+import pytest
+from infinity_context_adapters.postgres import build_async_engine, upgrade_schema
+from postgres_test_database import PostgresTestDatabase
+from postgres_versioned_schema_fixtures import install_versioned_schema_through
+from sqlalchemy import text
+
+
+def test_memory_outbox_width_upgrade_when_postgres_is_configured() -> None:
+    database_url = os.getenv("INFINITY_CONTEXT_TEST_POSTGRES_URL")
+    if not database_url:
+        pytest.skip("INFINITY_CONTEXT_TEST_POSTGRES_URL is not configured")
+    asyncio.run(_assert_width_upgrade(database_url))
+
+
+async def _assert_width_upgrade(database_url: str) -> None:
+    asyncpg = pytest.importorskip("asyncpg")
+    database = PostgresTestDatabase.from_url(
+        database_url, prefix="outbox_width_0060", asyncpg=asyncpg
+    )
+    await database.recreate()
+    try:
+        await install_versioned_schema_through(database, "0059_")
+        existing_id = "e" * 80
+        raw = await database.connect()
+        try:
+            await raw.execute(
+                "INSERT INTO memory_outbox "
+                "(event_type,aggregate_type,aggregate_id,payload_json,status,"
+                "attempt_count,next_attempt_at,created_at,updated_at) "
+                "VALUES ('probe.existing','locator_profile',$1,'{}','pending',0,"
+                "CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)",
+                existing_id,
+            )
+        finally:
+            await raw.close()
+
+        engine = build_async_engine(database.app_url)
+        try:
+            result = await upgrade_schema(engine)
+            assert result.applied == ("0060_memory_outbox_aggregate_id_width",)
+            aggregate_id = "p" * 120
+            async with engine.begin() as connection:
+                assert await connection.scalar(
+                    text(
+                        "SELECT character_maximum_length FROM information_schema.columns "
+                        "WHERE table_schema='public' AND table_name='memory_outbox' "
+                        "AND column_name='aggregate_id'"
+                    )
+                ) == 120
+                assert await connection.scalar(
+                    text("SELECT aggregate_id FROM memory_outbox WHERE aggregate_id=:id"),
+                    {"id": existing_id},
+                ) == existing_id
+                await connection.execute(
+                    text(
+                        "INSERT INTO memory_outbox "
+                        "(event_type,aggregate_type,aggregate_id,payload_json,status,"
+                        "attempt_count,next_attempt_at,created_at,updated_at) "
+                        "VALUES ('vector.upsert_locator_profile','locator_profile',:id,"
+                        "'{}','pending',0,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,"
+                        "CURRENT_TIMESTAMP)"
+                    ),
+                    {"id": aggregate_id},
+                )
+        finally:
+            await engine.dispose()
+    finally:
+        await database.drop()

tests/e2e/test_postgres_cleanup_plan_upgrade_e2e.py

@@ -54,8 +54,9 @@ async def _assert_pr57_history_upgrade(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
-            assert upgrade.current == "0059_locator_parent_lifecycle"
+            assert upgrade.current == "0060_memory_outbox_aggregate_id_width"
             assert (await upgrade_schema(engine)).applied == ()
             await _assert_cleanup_plan_schema(engine)
         finally:
@@ -110,6 +111,7 @@ async def _assert_cleanup_plan_upgrade(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
             await _assert_cleanup_plan_schema(engine)
             await _assert_projection_receipt_schema(engine)

tests/e2e/test_postgres_receipt_thread_scope_upgrade_e2e.py

@@ -83,7 +83,7 @@ async def _assert_receipt_and_thread_scope_upgrade(database_url: str) -> None:
         try:
             result = await upgrade_schema(engine)
             assert result.legacy_baseline is True
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
             await _assert_same_thread_relation_and_not_null_keys(engine)
             await _assert_thread_scope_fks_and_append_only_receipts(engine)
         finally:

tests/e2e/test_postgres_schema_upgrade_e2e.py

@@ -67,7 +67,7 @@ async def _assert_clean_and_legacy_upgrade(database_url: str) -> None:
             clean_results = await _run_concurrent_schema_upgrades(engine)
             clean = next(result for result in clean_results if result.applied)
             assert clean.legacy_baseline is False
-            assert clean.current == "0059_locator_parent_lifecycle"
+            assert clean.current == "0060_memory_outbox_aggregate_id_width"
             assert clean.applied[0] == "0001_core_facts"
             canonical_migration_count = len(_load_migrations())
             assert sorted(len(result.applied) for result in clean_results) == [
@@ -109,7 +109,7 @@ async def _assert_clean_and_legacy_upgrade(database_url: str) -> None:
             legacy = await upgrade_schema(engine)
             assert legacy.legacy_baseline is True
             assert legacy.applied[0].startswith("0023_")
-            assert legacy.current == "0059_locator_parent_lifecycle"
+            assert legacy.current == "0060_memory_outbox_aggregate_id_width"
             await _assert_head_schema(engine)
             await _assert_cross_scope_audit_reference_rejected(engine)
         finally:

tests/e2e/test_reconciliation_0049_populated_upgrade_postgres.py

@@ -93,7 +93,10 @@ async def _assert_populated_upgrade(database_url: str) -> None:
                     )
                 )
             result = await upgrade_schema(engine)
-            assert result.applied == ("0059_locator_parent_lifecycle",)
+            assert result.applied == (
+                "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
+            )
             async with engine.connect() as connection:
                 legacy_operation = (
                     await connection.execute(

tests/e2e/test_retrieval_profile_operator_receipts_postgres.py

@@ -40,7 +40,7 @@ async def _assert_durable_receipts(database_url: str) -> None:
     provenance = _provenance()
     try:
         upgraded = await upgrade_schema(engine)
-        assert upgraded.current == "0059_locator_parent_lifecycle"
+        assert upgraded.current == "0060_memory_outbox_aggregate_id_width"
         registry = PostgresRetrievalProfileRegistry(build_session_factory(engine))
         for operation in ("create", "rebuild", "attest", "activate"):
             key = f"lost-response-{operation}"

tests/e2e/test_strict_v4_document_execution_postgres.py

@@ -112,7 +112,7 @@ def phase(name: str) -> None:
         migration_engine = build_async_engine(database.app_url)
         try:
             result = await upgrade_schema(migration_engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
         finally:
             await migration_engine.dispose()
         phase("schema-upgraded")

tests/e2e/test_strict_v4_role_acl_upgrade_postgres.py

@@ -168,6 +168,7 @@ async def _scenario(database_url: str) -> None:
                 "0057_unmanaged_document_trigger_scope",
                 "0058_suggestion_server_thread_scope",
                 "0059_locator_parent_lifecycle",
+                "0060_memory_outbox_aggregate_id_width",
             )
         finally:
             await engine.dispose()
@@ -281,9 +282,9 @@ async def _non_schema_owner_scenario(database_url: str) -> None:
         engine = build_async_engine(migrator_database.app_url)
         try:
             result = await upgrade_schema(engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
             assert result.applied[0] == "0001_core_facts"
-            assert result.applied[-1] == "0059_locator_parent_lifecycle"
+            assert result.applied[-1] == "0060_memory_outbox_aggregate_id_width"
         finally:
             await engine.dispose()
 

tests/e2e/test_strict_v4_writer_fence_postgres.py

@@ -138,7 +138,7 @@ async def _assert_strict_v4_canonical_document_graph(database_url: str) -> None:
         migration_engine = build_async_engine(database.app_url)
         try:
             result = await upgrade_schema(migration_engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
         finally:
             await migration_engine.dispose()
 
@@ -307,7 +307,7 @@ async def _assert_strict_v4_canonical_fact_graph(database_url: str) -> None:
         migration_engine = build_async_engine(database.app_url)
         try:
             result = await upgrade_schema(migration_engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
         finally:
             await migration_engine.dispose()
 
@@ -562,7 +562,7 @@ async def _assert_strict_v4_writer_authority(database_url: str) -> None:
         engine = build_async_engine(database.app_url)
         try:
             result = await upgrade_schema(engine)
-            assert result.current == "0059_locator_parent_lifecycle"
+            assert result.current == "0060_memory_outbox_aggregate_id_width"
         finally:
             await engine.dispose()
         canonical_role = await database.create_runtime_role(

tests/migrations/test_locator_parent_lifecycle_migration.py

@@ -12,14 +12,14 @@
 
 def test_parent_lifecycle_repair_is_the_next_forward_only_migration() -> None:
     migrations = migration_runner._load_migrations()
-    assert migrations[-1].migration_id == "0059_locator_parent_lifecycle"
+    assert migrations[-2].migration_id == "0059_locator_parent_lifecycle"
     assert sum(item.migration_id == "0059_locator_parent_lifecycle" for item in migrations) == 1
 
 
 def test_pre_0059_binary_rejects_the_forward_only_history_row() -> None:
     migrations = migration_runner._load_migrations()
-    old_binary_migrations = migrations[:-1]
-    history = {item.migration_id: item.checksum for item in migrations}
+    old_binary_migrations = migrations[:-2]
+    history = {item.migration_id: item.checksum for item in migrations[:-1]}
 
     with pytest.raises(RuntimeError, match="Unknown applied PostgreSQL migration: 0059"):
         migration_runner._validate_history(old_binary_migrations, history)

tests/migrations/test_locator_profile_lifecycle_migration.py

@@ -7,7 +7,7 @@
 def test_profile_lifecycle_is_forward_only_after_published_0039() -> None:
     migrations = _load_migrations()
     ids = tuple(migration.migration_id for migration in migrations)
-    assert ids[-22:] == (
+    assert ids[-23:] == (
         "0039_locator_retrieval_attributes",
         "0040_locator_profile_lifecycle",
         "0041_locator_profile_attestation_fence",
@@ -30,6 +30,7 @@ def test_profile_lifecycle_is_forward_only_after_published_0039() -> None:
         "0057_unmanaged_document_trigger_scope",
         "0058_suggestion_server_thread_scope",
         "0059_locator_parent_lifecycle",
+        "0060_memory_outbox_aggregate_id_width",
     )
     sql = Path(__file__).resolve().parents[2] / (
         "packages/infinity_context_adapters/infinity_context_adapters/postgres/migrations/"

tests/migrations/test_locator_retrieval_migration.py

@@ -75,7 +75,7 @@ def test_published_ledger_prefix_continues_through_forward_locator_migration() -
 
     _validate_history(migrations, history)
 
-    assert migrations[-1].migration_id == "0059_locator_parent_lifecycle"
+    assert migrations[-1].migration_id == "0060_memory_outbox_aggregate_id_width"
 
 
 def test_published_locator_checksums_remain_upgrade_compatible() -> None:

tests/migrations/test_memory_outbox_aggregate_id_width.py

@@ -0,0 +1,52 @@
+from datetime import UTC, datetime
+from pathlib import Path
+
+from infinity_context_adapters.postgres import migration_runner
+from infinity_context_adapters.postgres.outbox_models import MemoryOutboxRow
+from sqlalchemy import create_engine
+
+MIGRATION = (
+    Path(__file__).resolve().parents[2]
+    / "packages/infinity_context_adapters/infinity_context_adapters/postgres/migrations"
+    / "0060_memory_outbox_aggregate_id_width.sql"
+)
+
+
+def test_0060_widens_only_the_outbox_aggregate_id() -> None:
+    migrations = migration_runner._load_migrations()
+    assert migrations[-2].migration_id == "0059_locator_parent_lifecycle"
+    assert migrations[-1].migration_id == "0060_memory_outbox_aggregate_id_width"
+    assert MIGRATION.read_text(encoding="utf-8") == (
+        "SET LOCAL lock_timeout = '5s';\n"
+        "SET LOCAL statement_timeout = '30s';\n"
+        "\n"
+        "ALTER TABLE public.memory_outbox\n"
+        "  ALTER COLUMN aggregate_id TYPE VARCHAR(120);\n"
+    )
+
+
+def test_outbox_model_accepts_a_120_character_aggregate_id_on_sqlite() -> None:
+    aggregate_id = "p" * 120
+    assert MemoryOutboxRow.__table__.c.aggregate_id.type.length == 120
+
+    engine = create_engine("sqlite://")
+    MemoryOutboxRow.__table__.create(engine)
+    now = datetime(2026, 9, 15, tzinfo=UTC)
+    with engine.begin() as connection:
+        connection.execute(
+            MemoryOutboxRow.__table__.insert(),
+            {
+                "event_type": "vector.upsert_locator_profile",
+                "aggregate_type": "locator_profile",
+                "aggregate_id": aggregate_id,
+                "payload_json": {},
+                "next_attempt_at": now,
+                "created_at": now,
+                "updated_at": now,
+            },
+        )
+        assert connection.scalar(
+            MemoryOutboxRow.__table__.select().with_only_columns(
+                MemoryOutboxRow.aggregate_id
+            )
+        ) == aggregate_id