feat(media): centralize resource URLs and persist refreshed image URLs

- Deduplicate message and history resource URLs using stable catalog references
- Persist renewed URLs with concurrent refresh protection
- Refresh OneBot images via get_msg intervals or NapCat rkey TTL
- Migrate historical payloads and cover SQLite and PostgreSQL behavior

ARCHITECTURE.md

@@ -122,7 +122,7 @@ Platform code lives under `pkg/platform/`.
 - Bots store exclusive Agent/Pipeline routes in `event_bindings` and independent plugin subscriptions in `plugin_processors` (`processor_uuid`, `enabled`). Subscriptions resolve event patterns from the installed Runner and fan out alongside the primary route. Configuration, state, debug and run logs belong to the reusable processor instance.
 - `sources/` contains adapter implementations. Each adapter subclasses `langbot_plugin.api.definition.abstract.platform.adapter.AbstractMessagePlatformAdapter` from the SDK.
 - Platform entities such as `MessageChain`, `Image`, `At`, `Voice`, and events come from `langbot-plugin-sdk`, not from this repo.
-- `pkg/platform/image_urls.py` renews image references in authorized monitoring/history responses and before Host model attempts without downloading or rewriting history. OneBot connection-local state in `adapters/aiocqhttp/image_urls.py` identifies the implementation lazily, reuses NapCat credentials until their returned TTL expires, and caches standard `get_msg` lookups for other implementations. Credentials and source mappings never cross connections or Workspace boundaries.
+- `pkg/platform/image_urls.py` renews image references in authorized monitoring/history responses and before Host model attempts without downloading. The shared resource URL catalog persists renewed URLs behind stable references. OneBot connection-local state in `adapters/aiocqhttp/image_urls.py` identifies the implementation lazily and reuses NapCat credentials until their returned TTL expires. Other implementations use standard `get_msg` when the URL's creation or last successful check is at least 30 minutes old. URLs can be shared across Workspaces; bot ownership checks, credentials and original message identities remain scoped to their connection.
 
 The platform layer should translate between external platform APIs and LangBot's shared message/event model. It should not contain LLM-provider logic or pipeline business logic.
 
@@ -275,6 +275,7 @@ Persistence is centered on `pkg/persistence/mgr.py`.
 - SQLite is the default database; PostgreSQL is supported.
 - Models live under `pkg/entity/persistence/`.
 - `pkg/persistence/media.py::MediaReferenceText` validates bound conversation payloads on ORM/Core writes while preserving the existing TEXT schema and string API. It rejects recognized structured inline images: retained images go through `MediaCache.externalize_chain_dump()` before serialization, while metadata-only snapshots and transcripts use `redact_inline_images()`. Reads and comparisons preserve historical values for incremental cleanup. Runtime/adapter messages, arbitrary diagnostic text, and opaque strings are outside this contract.
+- `pkg/persistence/resource_urls.py` interns structured resource URL fields in retained messages, history and monitoring at the typed SQL boundary, in the source transaction. `MediaReferenceJSON` preserves native JSON monitoring details. Payloads store opaque references to the global `resource_urls` catalog; typed SQL reads and comparisons expand them to current URLs, preserving API shapes and source-record authorization. Configuration URLs and URLs in free text are excluded. The URL hash has a unique index; canonical IDs and check timestamps are indexed. Successful refreshes overwrite the current URL and persist the check time even when standard `get_msg` returns the same URL. Converging URLs retain only ID redirects, with no old URL. PostgreSQL row locks and SQLite's immediate write transaction serialize concurrent refreshes; SQLite holds its write lock during the upstream call. Bounded in-memory aliases let reused Runner payloads resolve old URLs without retaining those URLs in the database. Migration `0041_resource_urls` normalizes existing payloads using their source creation timestamps before producers start; downgrade expands references before dropping the catalog.
 - Timezone-less SQL `DateTime` columns store UTC-naive values on both backends. Normalize aware values with `pkg/persistence/datetime_utils.py::as_naive_utc` at the bind boundary, including deadlines, leases, event times, and query/retention cutoffs; do not merely strip an offset. Existing naive rows represent UTC. Restore UTC awareness for application comparisons and epoch serialization. Runner ledger/event/transcript stores follow this existing schema contract without a data migration or changes to journal/tenant authorization.
 - Fresh schemas are created from current metadata, then Alembic migrations run to head. LangBot 4.x does not upgrade 3.x databases.
 - New schema changes use Alembic under `pkg/persistence/alembic/versions/`; there is no legacy migration chain in 4.x.

skills/skills/langbot-mcp-ops/SKILL.md

@@ -221,10 +221,11 @@ logs and action results. These times are not internal plugin profiling data.
 ## Unified execution monitoring
 
 Message record reads can renew temporary OneBot image URLs through the connected
-bot without downloading images or updating persisted payloads. Explicitly identified
+bot without downloading images. Persisted payloads reference a shared URL catalog;
+renewal overwrites its current URL so existing references read the new address. Explicitly identified
 NapCat connections reuse rkeys until their returned TTL expires (appid 1406 uses
 type 10; appid 1407 uses type 20). Other OneBot implementations use standard
-`get_msg` with a short per-message cache when original message and image identities
+`get_msg` at most once per URL per 30 minutes when original message and image identities
 are available. Disconnected peers, missing origins, or unsupported APIs leave the
 stored URL unchanged; returned URLs are not permanent archival links.
 

src/langbot/pkg/entity/persistence/agent_interaction.py

@@ -7,6 +7,7 @@ import datetime
 import sqlalchemy
 
 from .base import Base
+from ...persistence.media import MediaReferenceText
 
 
 class AgentInteraction(Base):
@@ -29,15 +30,15 @@ class AgentInteraction(Base):
     actor_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
 
     status = sqlalchemy.Column(sqlalchemy.String(50), nullable=False, index=True)
-    request_json = sqlalchemy.Column(sqlalchemy.Text, nullable=False)
+    request_json = sqlalchemy.Column(MediaReferenceText, nullable=False)
     delivery_target_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
-    delivery_result_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    delivery_result_json = sqlalchemy.Column(MediaReferenceText, nullable=True)
     replaces_interaction_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
     callback_token_hash = sqlalchemy.Column(sqlalchemy.String(64), nullable=False, unique=True, index=True)
 
     expires_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True, index=True)
     submitted_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
-    submission_json = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    submission_json = sqlalchemy.Column(MediaReferenceText, nullable=True)
     status_reason = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
 
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, default=datetime.datetime.utcnow)

src/langbot/pkg/entity/persistence/assistant.py

@@ -1,6 +1,7 @@
 import sqlalchemy as sa
 
 from .base import Base
+from ...persistence.media import MediaReferenceJSON
 
 
 class AssistantConversation(Base):
@@ -13,7 +14,7 @@ class AssistantConversation(Base):
     account_uuid = sa.Column(sa.String(36), sa.ForeignKey('users.uuid', ondelete='CASCADE'), nullable=False)
     revision = sa.Column(sa.Integer, nullable=False, default=0)
     status = sa.Column(sa.String(20), nullable=False, default='ready')
-    messages = sa.Column(sa.JSON, nullable=False, default=list)
+    messages = sa.Column(MediaReferenceJSON, nullable=False, default=list)
     error = sa.Column(sa.Text, nullable=True)
     model_name = sa.Column(sa.String(255), nullable=True)
     model_uuid = sa.Column(sa.String(36), nullable=True)

src/langbot/pkg/entity/persistence/monitoring.py

@@ -1,6 +1,6 @@
 import sqlalchemy
 
-from ...persistence.media import MediaReferenceText
+from ...persistence.media import MediaReferenceText, MediaReferenceJSON
 from .base import Base
 
 
@@ -107,7 +107,7 @@ class MonitoringLLMCall(Base):
     error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
     message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)  # Associated message ID
 
-    details = sqlalchemy.Column(sqlalchemy.JSON, nullable=True)
+    details = sqlalchemy.Column(MediaReferenceJSON, nullable=True)
 
     __table_args__ = (
         sqlalchemy.Index('ix_monitoring_llm_calls_workspace_timestamp', 'workspace_uuid', 'timestamp'),
@@ -137,8 +137,8 @@ class MonitoringToolCall(Base):
     pipeline_name = sqlalchemy.Column(sqlalchemy.String(255), nullable=False)
     session_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
     message_id = sqlalchemy.Column(sqlalchemy.String(255), nullable=True, index=True)
-    arguments = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
-    result = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
+    arguments = sqlalchemy.Column(MediaReferenceText, nullable=True)
+    result = sqlalchemy.Column(MediaReferenceText, nullable=True)
     error_message = sqlalchemy.Column(sqlalchemy.Text, nullable=True)
 
     __table_args__ = (

src/langbot/pkg/entity/persistence/pipeline.py

@@ -1,6 +1,7 @@
 import sqlalchemy
 
 from .base import Base
+from ...persistence.media import MediaReferenceJSON
 
 
 class LegacyPipeline(Base):
@@ -74,7 +75,7 @@ class PipelineRunRecord(Base):
     )
     started_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
     finished_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
-    result = sqlalchemy.Column(sqlalchemy.JSON, nullable=False)
+    result = sqlalchemy.Column(MediaReferenceJSON, nullable=False)
     knowledge_base_uuid = sqlalchemy.Column(sqlalchemy.String(255), nullable=True)
 
     __table_args__ = (

src/langbot/pkg/entity/persistence/resource_url.py

@@ -0,0 +1,26 @@
+"""Shared, stable identities for resource URLs embedded in conversation payloads."""
+
+import sqlalchemy as sa
+
+from .base import Base
+
+
+class ResourceURL(Base):
+    __tablename__ = 'resource_urls'
+
+    id = sa.Column(sa.String(36), primary_key=True)
+    url = sa.Column(sa.Text, nullable=True)
+    url_hash = sa.Column(sa.String(64), nullable=True, unique=True)
+    canonical_id = sa.Column(sa.String(36), sa.ForeignKey('resource_urls.id'), nullable=True, index=True)
+    created_at = sa.Column(sa.DateTime, nullable=False)
+    last_checked_at = sa.Column(sa.DateTime, nullable=True)
+    updated_at = sa.Column(sa.DateTime, nullable=False)
+
+    __table_args__ = (
+        sa.CheckConstraint(
+            '(canonical_id IS NULL AND url IS NOT NULL AND url_hash IS NOT NULL) '
+            'OR (canonical_id IS NOT NULL AND url IS NULL AND url_hash IS NULL)',
+            name='ck_resource_url_canonical',
+        ),
+        sa.Index('ix_resource_urls_last_checked', 'last_checked_at'),
+    )

src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py

@@ -0,0 +1,92 @@
+"""Normalize retained resource URLs behind stable shared references."""
+
+import datetime
+import json
+
+from alembic import op
+import sqlalchemy as sa
+
+revision = '0041_resource_urls'
+down_revision = '0040_session_summaries'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    from langbot.pkg.entity.persistence.base import Base
+    from langbot.pkg.entity.persistence.resource_url import ResourceURL
+    from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON
+    from langbot.pkg.persistence.resource_urls import create_postgres_expansion, intern, install
+
+    connection = op.get_bind()
+    ResourceURL.__table__.create(connection, checkfirst=True)
+    install(connection.engine)
+    if connection.dialect.name == 'postgresql':
+        create_postgres_expansion(connection)
+    inspector = sa.inspect(connection)
+    existing = set(inspector.get_table_names())
+    for table in sorted(Base.metadata.tables.values(), key=lambda table: table.name):
+        if table.name not in existing:
+            continue
+        retained = [
+            column.name for column in table.columns if isinstance(column.type, (MediaReferenceText, MediaReferenceJSON))
+        ]
+        json_columns = {column.name for column in table.columns if isinstance(column.type, MediaReferenceJSON)}
+        if not retained:
+            continue
+        # Reflection intentionally reads literal stored values without expansion.
+        source = sa.Table(table.name, sa.MetaData(), autoload_with=connection)
+        retained = [name for name in retained if name in source.c]
+        keys = list(source.primary_key.columns)
+        if len(keys) != 1:
+            raise RuntimeError(f'Resource URL migration requires one primary key: {table.name}')
+        key = keys[0]
+        cursor = None
+        stamp = source.c.get('created_at')
+        if stamp is None:
+            stamp = source.c.get('timestamp')
+        while True:
+            query = sa.select(source).order_by(key).limit(100)
+            if cursor is not None:
+                query = query.where(key > cursor)
+            rows = connection.execute(query).mappings().all()
+            if not rows:
+                break
+            for row in rows:
+                created = row[stamp.name] if stamp is not None else None
+                if created is not None and created.tzinfo is not None:
+                    created = created.astimezone(datetime.timezone.utc).replace(tzinfo=None)
+                updates = {}
+                for name in retained:
+                    raw = json.dumps(row[name]) if name in json_columns else row[name]
+                    converted = intern(connection, raw, created)
+                    updates[name] = json.loads(converted) if name in json_columns else converted
+                updates = {name: value for name, value in updates.items() if value != row[name]}
+                if updates:
+                    connection.execute(source.update().where(key == row[key.name]).values(**updates))
+            cursor = rows[-1][key.name]
+
+
+def downgrade():
+    from langbot.pkg.entity.persistence.base import Base
+    from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON
+    from langbot.pkg.persistence.resource_urls import uninstall
+
+    connection = op.get_bind()
+    existing = set(sa.inspect(connection).get_table_names())
+    for table in sorted(Base.metadata.tables.values(), key=lambda table: table.name):
+        if table.name not in existing:
+            continue
+        columns = {column['name'] for column in sa.inspect(connection).get_columns(table.name)}
+        for column in table.columns:
+            if column.name in columns and isinstance(column.type, (MediaReferenceText, MediaReferenceJSON)):
+                name = connection.dialect.identifier_preparer.quote(table.name)
+                field = connection.dialect.identifier_preparer.quote(column.name)
+                expanded = f'langbot_expand_resource_urls(CAST({field} AS TEXT))'
+                if isinstance(column.type, MediaReferenceJSON) and connection.dialect.name == 'postgresql':
+                    expanded = f'CAST({expanded} AS JSON)'
+                connection.exec_driver_sql(f'UPDATE {name} SET {field} = {expanded}')
+    op.drop_table('resource_urls')
+    if connection.dialect.name == 'postgresql':
+        connection.exec_driver_sql('DROP FUNCTION langbot_expand_resource_urls(text)')
+    uninstall(connection.engine)

src/langbot/pkg/persistence/media.py

@@ -5,7 +5,8 @@ from __future__ import annotations
 import json
 from typing import Any
 
-from sqlalchemy import Text
+from sqlalchemy import Text, JSON, cast
+from sqlalchemy.sql.elements import ColumnElement
 from sqlalchemy.engine import Dialect
 from sqlalchemy.sql.operators import OperatorType
 from sqlalchemy.types import TypeDecorator
@@ -23,9 +24,20 @@ class MediaReferenceText(TypeDecorator[str]):
     impl = Text
     cache_ok = True
 
+    class comparator_factory(Text.Comparator):
+        def operate(self, op, *other, **kw):
+            from .resource_urls import ExpandedResourceURLs
+
+            # ORM identity maps compare annotated copies of mapped columns.
+            if other and isinstance(other[0], ColumnElement) and self.expr.shares_lineage(other[0]):
+                return super().operate(op, *other, **kw)
+            # CAS and textual filters compare the same expanded value readers see.
+            return op(ExpandedResourceURLs(self.expr), *other, **kw)
+
     def process_bind_param(self, value: str | None, dialect: Dialect) -> str | None:
         if value is None:
             return None
+        reject_internal_references(value)
         try:
             payload = json.loads(value)
         except json.JSONDecodeError:
@@ -44,3 +56,37 @@ class MediaReferenceText(TypeDecorator[str]):
     def coerce_compared_value(self, op: OperatorType | None, value: Any) -> Text:
         # Legacy payloads must remain usable in searches and compare-and-swap updates.
         return self.impl
+
+    def column_expression(self, column):
+        from .resource_urls import ExpandedResourceURLs
+
+        return ExpandedResourceURLs(column)
+
+
+class MediaReferenceJSON(TypeDecorator):
+    """Preserve native JSON API values while interning structured resource fields."""
+
+    impl = JSON
+    cache_ok = True
+
+    def process_bind_param(self, value, dialect):
+        if value is not None:
+            reject_internal_references(json.dumps(value))
+        return value
+
+    def column_expression(self, column):
+        from .resource_urls import ExpandedResourceJSON
+
+        return ExpandedResourceJSON(cast(column, Text))
+
+
+def reject_internal_references(value):
+    """Only the transactional normalizer may introduce catalog references."""
+    from .resource_urls import REFERENCE
+
+    if REFERENCE.search(value):
+        raise ValueError('Internal resource references cannot be supplied as conversation content')
+
+
+# Register the catalog and its database hooks before engines are opened.
+from . import resource_urls as _resource_urls  # noqa: E402,F401

src/langbot/pkg/persistence/mgr.py

@@ -183,16 +183,25 @@ class PersistenceManager:
 
         if self.mode == PersistenceMode.CLOUD_RUNTIME:
             await self._validate_cloud_runtime()
+            from .resource_urls import install
+
+            install(engine.sync_engine)
             return
 
         if self.mode == PersistenceMode.RELEASE_MIGRATION:
             async with self._release_migration_lock():
                 await self._initialize_managed_schema()
                 await self._validate_release_schema()
+            from .resource_urls import install
+
+            install(engine.sync_engine)
             return
 
         await self._initialize_managed_schema()
         await self._enable_sqlite_foreign_keys_after_migration()
+        from .resource_urls import install
+
+        install(engine.sync_engine)
 
         if self.mode == PersistenceMode.OSS_COMPAT:
             await self.write_space_model_providers()

src/langbot/pkg/persistence/resource_urls.py

@@ -0,0 +1,391 @@
+"""Normalize resource URLs at the typed SQL boundary, within the source transaction.
+
+Only typed conversation payload writes are interned. Reads expand references in SQL so
+ORM and Core callers retain their existing string/JSON contracts. The catalog
+is shared; authorization continues to belong to the enclosing business record.
+"""
+
+import datetime
+import hashlib
+import json
+import re
+import uuid
+import sqlite3
+from collections import OrderedDict
+
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql, sqlite
+from sqlalchemy.ext.compiler import compiles
+from sqlalchemy.sql.functions import FunctionElement
+
+from ..entity.persistence.resource_url import ResourceURL
+
+
+PREFIX = 'langbot-resource://'
+REFERENCE = re.compile(r'langbot-resource://([a-f0-9-]{36})')
+FIELDS = frozenset({'url', 'image_url', 'audio_url', 'file_url', 'video_url', 'original_url'})
+
+
+def now():
+    return datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)
+
+
+def digest(url):
+    return hashlib.sha256(url.encode()).hexdigest()
+
+
+def fields(value):
+    if isinstance(value, list):
+        for child in value:
+            yield from fields(child)
+    elif isinstance(value, dict):
+        for key, child in value.items():
+            if key in FIELDS and isinstance(child, str) and child.startswith(('http://', 'https://', '/api/v1/files/')):
+                yield value, key, child
+            else:
+                yield from fields(child)
+
+
+def intern(connection, raw, created_at=None, identities_out=None):
+    """Intern resource fields synchronously in the current SQLAlchemy transaction."""
+    if not isinstance(raw, str):
+        return raw
+    try:
+        payload = json.loads(raw)
+    except ValueError:
+        return raw
+    nodes = list(fields(payload))
+    if not nodes:
+        return raw
+    table = ResourceURL.__table__
+    stamp = now()
+    urls = {url for _, _, url in nodes}
+    aliases = connection.engine._langbot_resource_aliases
+    known = {url: aliases[url] for url in urls if url in aliases}
+    valid = (
+        set(connection.execute(sa.select(table.c.id).where(table.c.id.in_(set(known.values())))).scalars())
+        if known
+        else set()
+    )
+    identities = {url: identity for url, identity in known.items() if identity in valid}
+    urls -= identities.keys()
+    hashes = dict(sorted((digest(url), url) for url in urls))
+    insert = {'sqlite': sqlite.insert, 'postgresql': postgresql.insert}[connection.dialect.name]
+    if hashes:
+        connection.execute(
+            insert(table)
+            .values(
+                [
+                    dict(id=str(uuid.uuid4()), url=url, url_hash=key, created_at=created_at or stamp, updated_at=stamp)
+                    for key, url in hashes.items()
+                ]
+            )
+            .on_conflict_do_nothing(index_elements=['url_hash'])
+        )
+    for identity, url, key in connection.execute(
+        sa.select(table.c.id, table.c.url, table.c.url_hash).where(table.c.url_hash.in_(hashes))
+    ):
+        if hashes[key] != url:
+            raise ValueError('Resource URL hash collision')
+        identities[url] = identity
+    if created_at is not None and identities:
+        # Migration order must not make an old shared URL appear newly created.
+        connection.execute(
+            table.update()
+            .where(table.c.id.in_(set(identities.values())), table.c.created_at > created_at)
+            .values(created_at=created_at)
+        )
+    for node, key, url in nodes:
+        node[key] = PREFIX + identities[url]
+        aliases[url] = identities[url]
+        aliases.move_to_end(url)
+    while len(aliases) > 8192:
+        aliases.popitem(last=False)
+    if identities_out is not None:
+        identities_out.update(identities)
+    return json.dumps(payload, ensure_ascii=False)
+
+
+class ExpandedResourceURLs(FunctionElement):
+    type = sa.Text()
+    inherit_cache = True
+
+
+class ExpandedResourceJSON(FunctionElement):
+    type = sa.JSON()
+    inherit_cache = True
+
+
+@compiles(ExpandedResourceJSON)
+def compile_json_expansion(element, compiler, **kw):
+    expression = compile_expansion(element, compiler, **kw)
+    if compiler.dialect.name == 'postgresql':
+        return f'CAST({expression} AS JSON)'
+    return expression
+
+
+@compiles(ExpandedResourceURLs)
+def compile_expansion(element, compiler, **kw):
+    source = compiler.process(list(element.clauses)[0], **kw)
+    if not getattr(compiler.dialect, '_langbot_resource_urls', False):
+        return source
+    return f'langbot_expand_resource_urls({source})'
+
+
+def install(engine):
+    """Enable normalization after schema initialization, before producers start."""
+    engine.dialect._langbot_resource_urls = True
+    if not hasattr(engine, '_langbot_resource_aliases'):
+        engine._langbot_resource_aliases = OrderedDict()
+    engine.clear_compiled_cache()
+    if not sa.event.contains(engine, 'before_cursor_execute', normalize_parameters):
+        sa.event.listen(engine, 'before_cursor_execute', normalize_parameters, retval=True)
+
+
+def uninstall(engine):
+    """Restore literal payload reads and writes after catalog downgrade."""
+    if sa.event.contains(engine, 'before_cursor_execute', normalize_parameters):
+        sa.event.remove(engine, 'before_cursor_execute', normalize_parameters)
+    engine.dialect._langbot_resource_urls = False
+    engine.clear_compiled_cache()
+    engine._langbot_resource_aliases.clear()
+
+
+def normalize_parameters(connection, cursor, statement, parameters, context, executemany):
+    """Replace only typed payload binds; never inspect arbitrary SQL or configuration."""
+    from .media import MediaReferenceText, MediaReferenceJSON
+
+    compiled = context.compiled
+    if compiled is None or not (context.isinsert or context.isupdate):
+        return statement, parameters
+    names = {
+        name
+        for bind, name in compiled.bind_names.items()
+        if isinstance(bind.type, (MediaReferenceText, MediaReferenceJSON))
+    }
+    if not names:
+        return statement, parameters
+
+    def normalize(row):
+        if isinstance(row, dict):
+            return {key: intern(connection, value) if key in names else value for key, value in row.items()}
+        positions = compiled.positiontup or []
+        if not positions:
+            return row
+        return tuple(
+            intern(connection, value) if positions[index % len(positions)] in names else value
+            for index, value in enumerate(row)
+        )
+
+    batched = executemany and getattr(compiled, '_insertmanyvalues', None) is None
+    return statement, [normalize(row) for row in parameters] if batched else normalize(parameters)
+
+
+def register_sqlite(dbapi, record):
+    """The UDF runs on aiosqlite's worker, including its catalog SELECTs."""
+    driver = getattr(dbapi, 'driver_connection', None)
+    raw_connection = driver._conn if hasattr(driver, '_conn') else dbapi
+    if not isinstance(raw_connection, sqlite3.Connection):
+        return
+
+    def expand(raw):
+        if not isinstance(raw, str) or PREFIX not in raw:
+            return raw
+
+        resolved = {}
+
+        def lookup(match):
+            identity = match.group(1)
+            if identity in resolved:
+                return resolved[identity]
+            visited = set()
+            while identity not in visited:
+                visited.add(identity)
+                row = raw_connection.execute(
+                    'SELECT url, canonical_id FROM resource_urls WHERE id=?', (identity,)
+                ).fetchone()
+                if row is None:
+                    raise ValueError('Missing resource URL')
+                if row[1] is None:
+                    value = json.dumps(row[0], ensure_ascii=False)[1:-1]
+                    for reference in visited:
+                        resolved[reference] = value
+                    return value
+                identity = row[1]
+            raise ValueError('Cyclic resource URL reference')
+
+        return REFERENCE.sub(lookup, raw)
+
+    dbapi.create_function('langbot_expand_resource_urls', 1, expand)
+
+
+POSTGRES_EXPANSION = r"""
+CREATE OR REPLACE FUNCTION public.langbot_expand_resource_urls(payload text)
+RETURNS text LANGUAGE plpgsql STABLE SECURITY INVOKER SET search_path = public, pg_temp AS $$
+DECLARE token text; identity text; target text; parent text; result text := payload; visited text[];
+BEGIN
+  IF payload IS NULL OR position('langbot-resource://' in payload) = 0 THEN RETURN payload; END IF;
+  FOR token IN SELECT DISTINCT m[1] FROM regexp_matches(payload, 'langbot-resource://([a-f0-9-]{36})', 'g') m LOOP
+    identity := token; visited := ARRAY[]::text[];
+    LOOP
+      IF identity = ANY(visited) THEN RAISE EXCEPTION 'Cyclic resource URL reference'; END IF;
+      visited := array_append(visited, identity);
+      SELECT url, canonical_id INTO target, parent FROM resource_urls WHERE id = identity;
+      IF NOT FOUND THEN RAISE EXCEPTION 'Missing resource URL'; END IF;
+      EXIT WHEN parent IS NULL;
+      identity := parent;
+    END LOOP;
+    result := replace(result, 'langbot-resource://' || token, substr(to_json(target)::text, 2, length(to_json(target)::text)-2));
+  END LOOP;
+  RETURN result;
+END $$;
+"""
+
+
+def catalog_created(table, connection, **kw):
+    if connection.dialect.name == 'postgresql':
+        create_postgres_expansion(connection)
+    install(connection.engine)
+
+
+def create_postgres_expansion(connection):
+    schema = connection.scalar(sa.select(sa.func.current_schema()))
+    quoted = connection.dialect.identifier_preparer.quote(schema)
+    connection.exec_driver_sql(POSTGRES_EXPANSION.replace('public', quoted))
+
+
+async def refresh_images(engine, peer, payload, *, is_current=None):
+    """Serialize refreshes on catalog identities and overwrite current URLs atomically."""
+    from ..platform.adapters.aiocqhttp.image_urls import image_nodes
+
+    nodes = list(image_nodes(payload))
+    if not nodes:
+        return
+
+    def connected():
+        return not peer.closed and (is_current is None or is_current())
+
+    table = ResourceURL.__table__
+    # Establish catalog identities for runtime-only inputs too. Existing history
+    # already has identities created in its original write or migration.
+    async with engine.begin() as db:
+        references = {}
+        await db.run_sync(lambda sync: intern(sync, json.dumps(payload), identities_out=references))
+    identities = set(references.values())
+    async with engine.connect() as db:
+        if engine.dialect.name == 'sqlite':
+            await db.exec_driver_sql('BEGIN IMMEDIATE')
+        else:
+            await db.begin()
+        try:
+            rows = (
+                (
+                    await db.execute(
+                        sa.select(table).where(table.c.id.in_(identities)).order_by(table.c.id).with_for_update()
+                    )
+                )
+                .mappings()
+                .all()
+            )
+            by_id = {row['id']: row for row in rows}
+            for identity in list(by_id):
+                row = by_id[identity]
+                while row['canonical_id'] is not None:
+                    row = (
+                        (await db.execute(sa.select(table).where(table.c.id == row['canonical_id']).with_for_update()))
+                        .mappings()
+                        .one()
+                    )
+                by_id[identity] = row
+            by_url = {node['url']: by_id[references[node['url']]] for node, _, _ in nodes}
+            # Capture origins even when renewal is not due or identification fails.
+            for node, message_id, image_id in nodes:
+                peer.image_urls.remember(by_url[node['url']]['url'], message_id, image_id)
+            # Identification and upstream credential handling remain connection-local.
+            if not connected() or not await peer.image_urls.identify():
+                return
+            napcat = peer.image_urls.app_name == 'NapCat.Onebot'
+            due = []
+            stamp = now()
+            for node, message_id, image_id in nodes:
+                row = by_url.get(node['url'])
+                if row is None:
+                    continue
+                node['url'] = row['url']
+                checked = row['last_checked_at'] or row['created_at']
+                if napcat or (stamp - checked).total_seconds() >= 1800:
+                    due.append((node, message_id, image_id, row))
+            if due:
+                # Keep Source/Image identities, including those recovered from events.
+                batch = [
+                    dict(message_id=message_id, image={'type': 'Image', 'image_id': image_id, 'url': node['url']})
+                    for node, message_id, image_id, _ in due
+                ]
+                old = [entry['image']['url'] for entry in batch]
+                completed = await peer.image_urls.refresh(batch)
+                if not completed:
+                    return
+                if not connected():
+                    return
+                stamp = now()
+                updated = set()
+                for (node, _, _, row), entry, previous in zip(due, batch, old):
+                    if previous not in completed:
+                        continue
+                    fresh = entry['image']['url']
+                    node['url'] = fresh
+                    if napcat and fresh == previous:
+                        continue
+                    if row['id'] in updated:
+                        continue
+                    updated.add(row['id'])
+                    values = dict(last_checked_at=stamp)
+                    if fresh != previous:
+                        fresh_hash = digest(fresh)
+                        # Lock the destination before merging identical current URLs.
+                        destination = (
+                            await db.execute(
+                                sa.select(table.c.id, table.c.url)
+                                .where(table.c.url_hash == fresh_hash)
+                                .with_for_update()
+                            )
+                        ).first()
+                        if destination is None:
+                            try:
+                                # Another source may discover the same renewed URL
+                                # concurrently. Let the unique index arbitrate it.
+                                async with db.begin_nested():
+                                    await db.execute(
+                                        table.update()
+                                        .where(table.c.id == row['id'])
+                                        .values(url=fresh, url_hash=fresh_hash, updated_at=stamp, **values)
+                                    )
+                                continue
+                            except sa.exc.IntegrityError as exc:
+                                if getattr(exc.orig, 'sqlstate', None) != '23505':
+                                    raise
+                                destination = (
+                                    await db.execute(
+                                        sa.select(table.c.id, table.c.url)
+                                        .where(table.c.url_hash == fresh_hash)
+                                        .with_for_update()
+                                    )
+                                ).one()
+                        if destination is not None and destination.id != row['id']:
+                            if destination.url != fresh:
+                                raise ValueError('Resource URL hash collision')
+                            values.update(url=None, url_hash=None, canonical_id=destination.id, updated_at=stamp)
+                            await db.execute(
+                                table.update().where(table.c.id == destination.id).values(last_checked_at=stamp)
+                            )
+                        else:
+                            values.update(url=fresh, url_hash=fresh_hash, updated_at=stamp)
+                    await db.execute(table.update().where(table.c.id == row['id']).values(**values))
+            await db.commit()
+        except BaseException:
+            await db.rollback()
+            raise
+
+
+sa.event.listen(sa.engine.Engine, 'connect', register_sqlite)
+sa.event.listen(ResourceURL.__table__, 'after_create', catalog_created)

src/langbot/pkg/persistence/tenant_uow.py

@@ -19,7 +19,7 @@ from sqlalchemy.dialects.postgresql.dml import OnConflictDoUpdate as PostgreSQLO
 from sqlalchemy.dialects.sqlite.dml import OnConflictDoNothing as SQLiteOnConflictDoNothing
 from sqlalchemy.dialects.sqlite.dml import OnConflictDoUpdate as SQLiteOnConflictDoUpdate
 
-from .media import MediaReferenceText
+from .media import MediaReferenceText, MediaReferenceJSON
 
 
 TENANT_SETTING = 'langbot.workspace_uuid'
@@ -295,7 +295,9 @@ def _validate_scoped_sql_type(
 
     if type(sql_type) in {Vector, HALFVEC}:
         return
-    if type(sql_type) is not MediaReferenceText and not type(sql_type).__module__.startswith('sqlalchemy.'):
+    if type(sql_type) not in {MediaReferenceText, MediaReferenceJSON} and not type(sql_type).__module__.startswith(
+        'sqlalchemy.'
+    ):
         raise ScopedSessionTransactionError('TenantUnitOfWork does not allow custom SQL types in public statements')
 
     for identifier_attribute in ('name', 'schema', 'collation'):
@@ -501,6 +503,14 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
                 )
 
         if isinstance(element, sqlalchemy.sql.functions.FunctionElement):
+            from .resource_urls import ExpandedResourceURLs
+
+            resource_expansion_allowed = (
+                type(element) is ExpandedResourceURLs
+                and len(list(element.clauses)) == 1
+                and isinstance(list(element.clauses)[0], sqlalchemy.Column)
+                and type(list(element.clauses)[0].type) is MediaReferenceText
+            )
             function_name = str(getattr(element, 'name', '')).casefold()
             package_names = tuple(getattr(element, 'packagenames', ()))
             generic_function_allowed = (
@@ -508,7 +518,9 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
                 and function_name in _ALLOWED_SCOPED_GENERIC_FUNCTIONS
             )
             builtin_function_allowed = _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES.get(function_name) is type(element)
-            if package_names or not (generic_function_allowed or builtin_function_allowed):
+            if package_names or not (
+                generic_function_allowed or builtin_function_allowed or resource_expansion_allowed
+            ):
                 raise ScopedSessionTransactionError(
                     f'TenantUnitOfWork does not allow SQL function {function_name or "<unknown>"!r}'
                 )

src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py

@@ -77,6 +77,7 @@ class OneBotImageURLs:
         self.keys = {}
         self.references = OrderedDict()
         self.messages = OrderedDict()
+        self.urls = OrderedDict()
         self.retry_after = 0
 
     def remember(self, url, message_id, image_id):
@@ -87,23 +88,37 @@ class OneBotImageURLs:
         while len(self.references) > 2048:
             self.references.popitem(last=False)
 
+    async def identify(self):
+        """Identify lazily with the same backoff as resource renewal."""
+        async with self.lock:
+            if self.connection.closed or time.monotonic() < self.retry_after:
+                return False
+            try:
+                await self._identify()
+                return True
+            except Exception:
+                self.retry_after = time.monotonic() + 30
+                raise
+
+    async def _identify(self):
+        if self.app_name is None:
+            info = await self.connection.call_action('get_version_info')
+            self.app_name = info.get('app_name', '')
+
     async def refresh(self, payload):
         nodes = list(image_nodes(payload))
         if not nodes or self.connection.closed:
-            return
+            return False
         for node, message_id, image_id in nodes:
             self.remember(node['url'], message_id, image_id)
         async with self.lock:
             if self.connection.closed or time.monotonic() < self.retry_after:
-                return
+                return False
             try:
-                if self.app_name is None:
-                    info = await self.connection.call_action('get_version_info')
-                    self.app_name = info.get('app_name', '')
+                await self._identify()
                 if self.app_name == 'NapCat.Onebot':
-                    await self._napcat(nodes)
-                else:
-                    await self._standard(nodes)
+                    return await self._napcat(nodes)
+                return await self._standard(nodes)
             except Exception:
                 # Failed/unsupported APIs must not break history reads or hammer the peer.
                 self.retry_after = time.monotonic() + 30
@@ -112,7 +127,7 @@ class OneBotImageURLs:
     async def _napcat(self, nodes):
         wanted = {kind for node, _, _ in nodes if (kind := nt_image_type(node['url'])) is not None}
         if not wanted:
-            return
+            return set()
         if any(self.keys.get(kind, ('', 0))[1] <= time.time() for kind in wanted):
             result = await self.connection.call_action('nc_get_rkey')
             keys = {}
@@ -129,10 +144,12 @@ class OneBotImageURLs:
             if not wanted.issubset(keys):
                 raise ValueError('NapCat returned no valid image credentials')
             self.keys.update(keys)
+        completed = set()
         for node, _, _ in nodes:
             kind = nt_image_type(node['url'])
             if kind not in wanted:
                 continue
+            completed.add(node['url'])
             parsed = urlsplit(node['url'])
             parts = parsed.query.split('&')
             # Preserve every other parameter byte-for-byte, including signed resource IDs.
@@ -143,10 +160,17 @@ class OneBotImageURLs:
                 for part in parts
             ]
             node['url'] = urlunsplit(parsed._replace(query='&'.join(parts)))
+        return completed
 
     async def _standard(self, nodes):
+        completed = set()
         for node, _, _ in nodes:
             old_url = node['url']
+            refreshed = self.urls.get(old_url)
+            if refreshed is not None and refreshed[1] > time.monotonic():
+                node['url'] = refreshed[0]
+                completed.add(old_url)
+                continue
             ref = self.references.get(old_url)
             if ref is None:
                 continue
@@ -160,12 +184,19 @@ class OneBotImageURLs:
                         data = segment.data
                         if data.get('file') and remote_image_url(data.get('url')):
                             images[str(data['file'])] = data['url']
-                cached = (images, time.monotonic() + 60)
+                cached = (images, time.monotonic() + 1800)
                 self.messages[str(message_id)] = cached
                 self.messages.move_to_end(str(message_id))
                 while len(self.messages) > 256:
                     self.messages.popitem(last=False)
             new_url = cached[0].get(image_id)
             if new_url:
+                completed.add(old_url)
                 self.remember(new_url, message_id, image_id)
+                deadline = cached[1]
+                self.urls[old_url] = (new_url, deadline)
+                self.urls[new_url] = (new_url, deadline)
+                while len(self.urls) > 2048:
+                    self.urls.popitem(last=False)
                 node['url'] = new_url
+        return completed

src/langbot/pkg/platform/adapters/aiocqhttp/transport.py

@@ -102,6 +102,7 @@ class OneBotConnection:
         self.image_urls.keys.clear()
         self.image_urls.references.clear()
         self.image_urls.messages.clear()
+        self.image_urls.urls.clear()
         for future in self.pending.values():
             if not future.done():
                 future.set_exception(ApiNotAvailable('OneBot connection closed'))

src/langbot/pkg/platform/image_urls.py

@@ -1,4 +1,4 @@
-"""Refresh image references at authorized read/use boundaries, without persistence."""
+"""Refresh image references at authorized read/use boundaries."""
 
 import copy
 import json
@@ -21,7 +21,14 @@ async def refresh_image_payload(ap, workspace_uuid, bot_id, payload):
         return payload
     result = copy.deepcopy(payload)
     try:
-        await connection.image_urls.refresh(result)
+        from ..persistence.resource_urls import refresh_images
+
+        await refresh_images(
+            ap.persistence_mgr.get_db_engine(),
+            connection,
+            result,
+            is_current=lambda: bot.adapter.bot.connection is connection,
+        )
         if bot.adapter.bot.connection is not connection or connection.closed:
             return payload
         return result

tests/integration/persistence/test_monitoring_postgres.py

@@ -9,9 +9,11 @@ from __future__ import annotations
 
 import datetime
 import logging
+import json
 import os
 import uuid
 from types import SimpleNamespace
+from unittest.mock import AsyncMock
 
 import pytest
 import pytest_asyncio
@@ -365,6 +367,43 @@ async def test_rls_is_enforced_without_application_workspace_predicates(service,
         await service.ap.persistence_mgr.execute_async(sa.select(models.MonitoringMessage.id))
 
 
+async def test_shared_resource_urls_preserve_cloud_workspace_authorization(service, cloud_database):
+    from langbot.pkg.entity.persistence.resource_url import ResourceURL
+    from langbot.pkg.persistence.resource_urls import refresh_images
+    from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs
+
+    _, admin = cloud_database
+    old, fresh = 'https://example.test/old-image', 'https://example.test/current-image'
+    chain = [{'type': 'Source', 'id': 42}, {'type': 'Image', 'image_id': 'one.jpg', 'url': old}]
+    records = {}
+    for workspace in (WORKSPACE_A, WORKSPACE_B):
+        records[workspace] = await service.record_message(
+            _context(workspace), **RESOURCE, session_id='shared-url', message_content=json.dumps(chain)
+        )
+    async with admin.begin() as db:
+        assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1
+        await db.execute(sa.update(ResourceURL).values(created_at=datetime.datetime(2000, 1, 1)))
+
+    async def action(name, **params):
+        if name == 'get_version_info':
+            return {'app_name': 'standard'}
+        assert name == 'get_msg' and params == {'message_id': 42}
+        return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': fresh}}]}
+
+    peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action))
+    peer.image_urls = OneBotImageURLs(peer)
+    # Exercise catalog writes using the actual non-superuser runtime role.
+    await refresh_images(service.ap.persistence_mgr.get_db_engine(), peer, chain)
+    for workspace in (WORKSPACE_A, WORKSPACE_B):
+        details = await _read(service, 'get_message_details', _context(workspace), records[workspace])
+        assert fresh in details['message']['message_content']
+        other = WORKSPACE_B if workspace == WORKSPACE_A else WORKSPACE_A
+        assert (await _read(service, 'get_message_details', _context(workspace), records[other]))['found'] is False
+    async with admin.connect() as db:
+        assert await db.scalar(sa.select(ResourceURL.url)) == fresh
+        assert old not in str((await db.execute(sa.text('SELECT message_content FROM monitoring_messages'))).all())
+
+
 async def test_traffic_series_aggregates_all_rows_under_cloud_rls(service):
     import datetime
     from langbot.pkg.api.http.service.monitoring_traffic import get_traffic_series

tests/unit_tests/api/service/test_monitoring_executions.py

@@ -84,12 +84,13 @@ async def test_event_properties_survive_detail_snapshot_fallback(engine, service
 @pytest.mark.asyncio
 @pytest.mark.parametrize('napcat', [True, False])
 @pytest.mark.parametrize('source', ['agent', 'pipeline'])
-async def test_execution_details_refresh_existing_images_without_persisting(engine, service, napcat, source):
+async def test_execution_details_refresh_existing_images_through_shared_catalog(engine, service, napcat, source):
     import json
     from unittest.mock import AsyncMock
     import sqlalchemy as sa
     from langbot.pkg.entity.persistence.event_log import EventLog
     from langbot.pkg.entity.persistence.transcript import Transcript
+    from langbot.pkg.entity.persistence.resource_url import ResourceURL
     from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs
 
     old = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=existing&rkey=expired'
@@ -140,6 +141,9 @@ async def test_execution_details_refresh_existing_images_without_persisting(engi
         )
         await session.commit()
 
+    async with engine.begin() as db:
+        await db.execute(sa.update(ResourceURL).values(created_at=_dt(10)))
+
     async def action(name, **params):
         if name == 'get_version_info':
             return {'app_name': 'NapCat.Onebot' if napcat else 'standard-peer'}
@@ -170,8 +174,10 @@ async def test_execution_details_refresh_existing_images_without_persisting(engi
     await service.get_execution_detail(WORKSPACE, source, identity, section='conversation')
     assert peer.call_action.await_count == 2
     async with engine.connect() as db:
-        assert old in (await db.execute(sa.select(EventLog.input_json))).scalar_one()
-        assert old in (await db.execute(sa.select(Transcript.content_json))).scalar_one()
+        assert new in (await db.execute(sa.select(EventLog.input_json))).scalar_one()
+        assert new in (await db.execute(sa.select(Transcript.content_json))).scalar_one()
+        raw = (await db.execute(sa.text('SELECT input_json FROM event_log'))).scalar_one()
+        assert 'langbot-resource://' in raw and old not in raw and new not in raw
 
 
 def test_event_content_preserves_conflicting_custom_input_fields():

tests/unit_tests/persistence/test_resource_urls.py

@@ -0,0 +1,367 @@
+"""URL catalog storage, transactional refresh and migration contracts."""
+
+import asyncio
+import datetime as dt
+import importlib
+import json
+import os
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+
+from langbot.pkg.entity.persistence.resource_url import ResourceURL
+from langbot.pkg.entity.persistence.event_log import EventLog
+from langbot.pkg.entity.persistence.monitoring import MonitoringLLMCall
+from langbot.pkg.persistence.media import MediaReferenceText, MediaReferenceJSON
+from langbot.pkg.persistence.resource_urls import digest, refresh_images
+from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs
+
+
+OLD = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=one&rkey=old'
+NEW = OLD.replace('rkey=old', 'rkey=new')
+
+
+def chain(url=OLD, message_id=42):
+    return [{'type': 'Source', 'id': message_id}, {'type': 'Image', 'image_id': 'one.jpg', 'url': url}]
+
+
+@pytest.fixture(params=['sqlite', 'postgresql'])
+async def database(tmp_path, request):
+    admin = None
+    schema = None
+    if request.param == 'postgresql':
+        url = os.environ.get('TEST_POSTGRES_URL')
+        if not url:
+            pytest.skip('TEST_POSTGRES_URL not set')
+        admin = create_async_engine(url)
+        schema = 'resource_' + uuid.uuid4().hex
+        async with admin.begin() as db:
+            await db.execute(sa.schema.CreateSchema(schema))
+        engine = create_async_engine(url, connect_args={'server_settings': {'search_path': schema}})
+    else:
+        engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "catalog.db"}')
+    try:
+        async with engine.begin() as db:
+            await db.run_sync(ResourceURL.__table__.create)
+            await db.run_sync(EventLog.__table__.create)
+        yield engine
+    finally:
+        await engine.dispose()
+        if admin is not None:
+            async with admin.begin() as db:
+                await db.execute(sa.schema.DropSchema(schema, cascade=True))
+            await admin.dispose()
+
+
+async def append(db, identity, url=OLD):
+    await db.execute(
+        sa.insert(EventLog).values(
+            event_id=identity,
+            event_type='message.received',
+            source='platform',
+            workspace_id='workspace',
+            bot_id='bot',
+            input_json=json.dumps(chain(url)),
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_dedup_expansion_and_rollback(database):
+    async with database.begin() as db:
+        await append(db, 'a')
+        await append(db, 'b')
+    async with database.connect() as db:
+        assert (await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL))) == 1
+        stored = (await db.execute(sa.text('SELECT input_json FROM event_log'))).scalars().all()
+        assert stored[0] == stored[1] and OLD not in stored[0]
+        assert OLD in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a'))
+        # The original string remains usable for compare-and-swap updates.
+        value = await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a'))
+        result = await db.execute(
+            sa.update(EventLog)
+            .where(EventLog.event_id == 'a', EventLog.input_json == value)
+            .values(input_json=json.dumps(chain(NEW)))
+        )
+        assert result.rowcount == 1
+        await db.rollback()
+    async with database.connect() as db:
+        assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1
+
+
+@pytest.mark.asyncio
+async def test_untrusted_content_cannot_inject_catalog_ids(database):
+    async with database.begin() as db:
+        await append(db, 'a')
+        identity = await db.scalar(sa.select(ResourceURL.id))
+    with pytest.raises(sa.exc.StatementError, match='Internal resource references'):
+        async with database.begin() as db:
+            await append(db, 'b', 'langbot-resource://' + identity)
+
+
+@pytest.mark.asyncio
+async def test_orm_update_keeps_loaded_payload_synchronized(database):
+    async with database.begin() as db:
+        await append(db, 'a')
+    async with AsyncSession(database, expire_on_commit=False) as session:
+        record = await session.scalar(sa.select(EventLog))
+        changed = json.dumps(chain(NEW))
+        await session.execute(sa.update(EventLog).where(EventLog.event_id == 'a').values(input_json=changed))
+        await session.commit()
+        assert record.input_json == changed
+        await session.refresh(record)
+        assert NEW in record.input_json
+
+
+@pytest.mark.asyncio
+async def test_bulk_and_native_json_preserve_api_shapes(database):
+    meta = sa.MetaData()
+    sample = sa.Table(
+        'sample',
+        meta,
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('body', MediaReferenceText),
+        sa.Column('detail', MediaReferenceJSON),
+    )
+    async with database.begin() as db:
+        await db.run_sync(meta.create_all)
+        await db.execute(
+            sample.insert(),
+            [dict(id=i, body=json.dumps(chain()), detail={'url': OLD, 'flag': False, 'number': 0}) for i in range(3)],
+        )
+        rows = (await db.execute(sa.select(sample))).mappings().all()
+        assert all(OLD in row['body'] and row['detail'] == {'url': OLD, 'flag': False, 'number': 0} for row in rows), (
+            rows
+        )
+        assert await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL)) == 1
+        assert (
+            len(
+                (
+                    await db.execute(
+                        sample.insert().returning(sample.c.id),
+                        [dict(id=i, body=json.dumps(chain()), detail={'url': OLD}) for i in range(3, 6)],
+                    )
+                ).all()
+            )
+            == 3
+        )
+        assert all(OLD in value for value in (await db.execute(sa.select(sample.c.body))).scalars())
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('catalog_url', [OLD, NEW])
+async def test_fresh_image_remembers_origin_for_later_model_history(database, catalog_url):
+    async with database.begin() as db:
+        await append(db, 'a')
+        await db.execute(sa.update(ResourceURL).values(url=catalog_url, url_hash=digest(catalog_url)))
+
+    renewed = NEW + '&renewed=1'
+
+    async def action(name, **params):
+        if name == 'get_version_info':
+            return {'app_name': 'standard'}
+        assert name == 'get_msg' and params == {'message_id': 42}
+        return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': renewed}}]}
+
+    peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action))
+    peer.image_urls = OneBotImageURLs(peer)
+    initial = chain()
+    await refresh_images(database, peer, initial)
+    assert initial[1]['url'] == catalog_url
+    assert peer.image_urls.references[catalog_url] == (42, 'one.jpg')
+    assert not any(call.args[0] == 'get_msg' for call in peer.call_action.call_args_list)
+
+    async with database.begin() as db:
+        assert await db.scalar(sa.select(ResourceURL.last_checked_at)) is None
+        await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1)))
+    history = [{'type': 'image_url', 'image_url': {'url': catalog_url}}]
+    await refresh_images(database, peer, history)
+    assert history[0]['image_url']['url'] == renewed
+    assert sum(call.args[0] == 'get_msg' for call in peer.call_action.call_args_list) == 1
+    async with database.connect() as db:
+        assert renewed in await db.scalar(sa.select(EventLog.input_json))
+        assert await db.scalar(sa.select(ResourceURL.last_checked_at)) is not None
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('converge', [False, True])
+async def test_refresh_shared_identity_concurrency_and_half_hour(database, converge):
+    async with database.begin() as db:
+        await append(db, 'a')
+        if converge:
+            await append(db, 'b', NEW)
+        await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1)))
+
+    async def action(name, **params):
+        await asyncio.sleep(0)
+        if name == 'get_version_info':
+            return {'app_name': 'other'}
+        assert name == 'get_msg'
+        return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': NEW}}]}
+
+    def peer():
+        result = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action))
+        result.image_urls = OneBotImageURLs(result)
+        return result
+
+    peers = [peer(), peer()]
+    batches = [chain(message_id=42), chain(message_id=43)]
+    await asyncio.gather(*(refresh_images(database, p, batch) for p, batch in zip(peers, batches)))
+    assert all(batch[1]['url'] == NEW for batch in batches)
+    assert sum(call.args[0] == 'get_msg' for p in peers for call in p.call_action.call_args_list) == 1
+    async with database.connect() as db:
+        assert NEW in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'a'))
+        assert (
+            await db.scalar(sa.select(sa.func.count()).select_from(ResourceURL).where(ResourceURL.url.is_not(None)))
+            == 1
+        )
+        assert OLD not in str((await db.execute(sa.select(ResourceURL.url))).scalars().all())
+    # A fresh connection still reuses the persisted last-check time.
+    replacement = peer()
+    await refresh_images(database, replacement, chain(NEW))
+    assert not any(call.args[0] == 'get_msg' for call in replacement.call_action.call_args_list)
+    async with database.begin() as db:
+        await db.execute(sa.update(ResourceURL).values(last_checked_at=dt.datetime(2000, 1, 1)))
+    await refresh_images(database, replacement, chain(NEW))
+    assert any(call.args[0] == 'get_msg' for call in replacement.call_action.call_args_list)
+
+
+@pytest.mark.asyncio
+async def test_distinct_urls_can_converge_concurrently(database):
+    old_urls = [OLD, OLD + '&variant=two']
+    async with database.begin() as db:
+        for index, url in enumerate(old_urls):
+            await append(db, str(index), url)
+        await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1)))
+    ready = asyncio.Event()
+    entered = 0
+
+    async def action(name, **params):
+        nonlocal entered
+        if name == 'get_version_info':
+            return {'app_name': 'standard'}
+        entered += 1
+        if database.dialect.name == 'postgresql':
+            if entered == 2:
+                ready.set()
+            await asyncio.wait_for(ready.wait(), 5)
+        return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': NEW}}]}
+
+    peers = [SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action)) for _ in old_urls]
+    for peer in peers:
+        peer.image_urls = OneBotImageURLs(peer)
+    await asyncio.gather(*(refresh_images(database, peer, chain(url)) for peer, url in zip(peers, old_urls)))
+    async with database.connect() as db:
+        assert all(NEW in value for value in (await db.execute(sa.select(EventLog.input_json))).scalars())
+        assert (await db.execute(sa.select(ResourceURL.url).where(ResourceURL.url.is_not(None)))).scalars().all() == [
+            NEW
+        ]
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('outcome', ['unchanged', 'missing_origin', 'failure'])
+async def test_only_successful_checks_advance_the_persisted_interval(database, outcome):
+    async with database.begin() as db:
+        await append(db, 'a')
+        await db.execute(sa.update(ResourceURL).values(created_at=dt.datetime(2000, 1, 1)))
+
+    async def action(name, **params):
+        if name == 'get_version_info':
+            return {'app_name': 'standard'}
+        if outcome == 'failure':
+            raise RuntimeError('upstream unavailable')
+        return {'message': [{'type': 'image', 'data': {'file': 'one.jpg', 'url': OLD}}]}
+
+    peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action))
+    peer.image_urls = OneBotImageURLs(peer)
+    payload = chain() if outcome != 'missing_origin' else [{'type': 'image_url', 'image_url': {'url': OLD}}]
+    if outcome == 'failure':
+        with pytest.raises(RuntimeError, match='upstream unavailable'):
+            await refresh_images(database, peer, payload)
+    else:
+        await refresh_images(database, peer, payload)
+    async with database.connect() as db:
+        assert (await db.scalar(sa.select(ResourceURL.last_checked_at)) is not None) == (outcome == 'unchanged')
+
+
+@pytest.mark.asyncio
+async def test_napcat_uses_key_ttl_instead_of_the_standard_interval(database):
+    async with database.begin() as db:
+        await append(db, 'a')
+    generation = 0
+
+    async def action(name, **params):
+        nonlocal generation
+        if name == 'get_version_info':
+            return {'app_name': 'NapCat.Onebot'}
+        assert name == 'nc_get_rkey'
+        generation += 1
+        return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': f'key{generation}'}]
+
+    peer = SimpleNamespace(closed=False, call_action=AsyncMock(side_effect=action))
+    peer.image_urls = OneBotImageURLs(peer)
+    await refresh_images(database, peer, chain())
+    async with database.connect() as db:
+        checked = await db.scalar(sa.select(ResourceURL.last_checked_at))
+        assert 'rkey=key1' in await db.scalar(sa.select(EventLog.input_json))
+    await refresh_images(database, peer, chain())
+    assert generation == 1
+    async with database.connect() as db:
+        assert await db.scalar(sa.select(ResourceURL.last_checked_at)) == checked
+    peer.image_urls.keys[20] = ('key1', 0)
+    await refresh_images(database, peer, chain())
+    assert generation == 2
+    async with database.connect() as db:
+        assert 'rkey=key2' in await db.scalar(sa.select(EventLog.input_json))
+
+
+@pytest.mark.asyncio
+async def test_migration_preserves_source_creation_and_downgrade(database):
+    migration = importlib.import_module('langbot.pkg.persistence.alembic.versions.0041_resource_urls')
+    async with database.begin() as db:
+        await append(db, 'newer')
+        historical = sa.Table(
+            'monitoring_llm_calls',
+            sa.MetaData(),
+            sa.Column('id', sa.String, primary_key=True),
+            sa.Column('details', sa.JSON),
+        )
+        await db.run_sync(historical.create)
+        await db.execute(historical.insert().values(id='old', details={'url': OLD, 'flag': False}))
+        # Simulate a historical write before catalog normalization existed.
+        await db.execute(
+            sa.text(
+                'INSERT INTO event_log (event_id,event_type,source,input_json,created_at) '
+                "VALUES ('old','message.received','platform',:payload,'2000-01-01 00:00:00')"
+            ),
+            {'payload': json.dumps(chain())},
+        )
+
+        def upgrade(sync):
+            with Operations.context(MigrationContext.configure(sync)):
+                migration.upgrade()
+                migration.upgrade()
+
+        await db.run_sync(upgrade)
+        assert OLD in await db.scalar(sa.select(EventLog.input_json))
+        assert (await db.scalar(sa.select(ResourceURL.created_at))).year == 2000
+        assert OLD not in await db.scalar(sa.text('SELECT input_json FROM event_log'))
+        assert await db.scalar(sa.select(MonitoringLLMCall.details)) == {'url': OLD, 'flag': False}
+        assert 'langbot-resource://' in str(await db.scalar(sa.select(historical.c.details)))
+
+        def downgrade(sync):
+            with Operations.context(MigrationContext.configure(sync)):
+                migration.downgrade()
+
+        await db.run_sync(downgrade)
+        assert OLD in await db.scalar(sa.text('SELECT input_json FROM event_log'))
+        assert await db.scalar(sa.select(historical.c.details)) == {'url': OLD, 'flag': False}
+        # The same engine can write and read literal payloads after downgrade.
+        await append(db, 'after', NEW)
+        assert NEW in await db.scalar(sa.select(EventLog.input_json).where(EventLog.event_id == 'after'))

tests/unit_tests/platform/test_onebot_image_urls.py

@@ -2,13 +2,17 @@
 
 import asyncio
 import copy
+import datetime
 import json
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 from urllib.parse import parse_qs, urlsplit
 
 import pytest
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import create_async_engine
 
+from langbot.pkg.entity.persistence.resource_url import ResourceURL
 from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs, nt_image_type
 from langbot.pkg.platform.image_urls import refresh_image_payload, refresh_model_images, refresh_history_items
 from langbot_plugin.api.entities.builtin.provider.message import Message
@@ -105,6 +109,7 @@ async def test_other_implementation_uses_original_id_and_caches_messages():
     await peer.image_urls.refresh(model_input)
     assert peer.call_action.await_count == 2
     peer.image_urls.messages.clear()
+    peer.image_urls.urls.clear()
     await peer.image_urls.refresh(model_input)
     assert peer.call_action.await_count == 3
 
@@ -128,21 +133,34 @@ async def test_aggregated_messages_keep_each_original_source():
     assert peer.call_action.await_count == 3
 
 
-def application(peer, workspace='workspace'):
+@pytest.fixture
+async def catalog():
+    engine = create_async_engine('sqlite+aiosqlite:///:memory:')
+    async with engine.begin() as db:
+        await db.run_sync(ResourceURL.__table__.create)
+    yield engine
+    await engine.dispose()
+
+
+def application(peer, engine, workspace='workspace'):
     bot = SimpleNamespace(
         workspace_uuid=workspace,
         enable=True,
         bot_entity=SimpleNamespace(adapter='onebot-v11'),
         adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)),
     )
-    ap = SimpleNamespace(platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)), logger=Mock())
+    ap = SimpleNamespace(
+        platform_mgr=SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot)),
+        persistence_mgr=SimpleNamespace(get_db_engine=lambda: engine),
+        logger=Mock(),
+    )
     return ap, bot
 
 
 @pytest.mark.asyncio
-async def test_scope_failure_backoff_and_no_persistence():
+async def test_scope_failure_backoff_preserves_current_url(catalog):
     peer = connection(lambda *_args, **_kwargs: {'app_name': 'NapCat.Onebot'})
-    ap, bot = application(peer)
+    ap, bot = application(peer, catalog)
     original = payload(url())
     saved = copy.deepcopy(original)
     assert await refresh_image_payload(ap, 'foreign', 'bot', original) is original
@@ -158,14 +176,14 @@ async def test_scope_failure_backoff_and_no_persistence():
 
 
 @pytest.mark.asyncio
-async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages():
+async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages(catalog):
     async def action(name, **_params):
         if name == 'get_version_info':
             return {'app_name': 'NapCat.Onebot'}
         return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': '&rkey=current'}]
 
     peer = connection(action)
-    ap, _ = application(peer)
+    ap, _ = application(peer, catalog)
     messages = [Message(role='user', content=[{'type': 'image_url', 'image_url': {'url': url()}}])]
     context = SimpleNamespace(workspace_uuid='workspace')
     query = SimpleNamespace(bot_uuid='bot')
@@ -180,7 +198,7 @@ async def test_model_reuse_refreshes_with_current_keys_without_mutating_messages
 
 
 @pytest.mark.asyncio
-async def test_disconnected_peer_cannot_return_refreshed_result():
+async def test_disconnected_peer_cannot_return_refreshed_result(catalog):
     async def action(name, **_params):
         if name == 'get_version_info':
             return {'app_name': 'NapCat.Onebot'}
@@ -188,18 +206,18 @@ async def test_disconnected_peer_cannot_return_refreshed_result():
         return [{'type': 20, 'time': 1, 'ttl': 10**12, 'rkey': 'new'}]
 
     peer = connection(action)
-    ap, bot = application(peer)
+    ap, bot = application(peer, catalog)
     original = payload(url())
     assert await refresh_image_payload(ap, 'workspace', 'bot', original) is original
+    async with catalog.connect() as db:
+        assert await db.scalar(sa.select(ResourceURL.url)) == url()
 
 
 @pytest.mark.asyncio
-async def test_history_recovers_scoped_origins_without_rewriting_event_facts():
-    import sqlalchemy as sa
-    from sqlalchemy.ext.asyncio import create_async_engine
+async def test_history_recovers_scoped_origins_and_updates_shared_url(catalog):
     from langbot.pkg.entity.persistence.event_log import EventLog
 
-    engine = create_async_engine('sqlite+aiosqlite:///:memory:')
+    engine = catalog
     original = payload('https://example.test/old')
     metadata = json.dumps({'input_event': {'message_chain': original}})
     async with engine.begin() as db:
@@ -214,6 +232,7 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts():
                 metadata_json=metadata,
             )
         )
+        await db.execute(sa.update(ResourceURL).values(created_at=datetime.datetime(2000, 1, 1)))
 
     async def execute(statement):
         async with engine.connect() as db:
@@ -226,8 +245,8 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts():
         return {'message': [{'type': 'image', 'data': {'file': '0.jpg', 'url': 'https://example.test/new'}}]}
 
     peer = connection(action)
-    ap, _ = application(peer)
-    ap.persistence_mgr = SimpleNamespace(execute_async=execute)
+    ap, _ = application(peer, catalog)
+    ap.persistence_mgr.execute_async = execute
     item = {
         'bot_id': 'bot',
         'workspace_id': 'workspace',
@@ -240,7 +259,9 @@ async def test_history_recovers_scoped_origins_without_rewriting_event_facts():
         await refresh_history_items(ap, 'workspace', 'bot', [item])
         assert item['content_json']['content'][0]['image_url']['url'] == 'https://example.test/new'
         result = await execute(sa.select(EventLog.metadata_json))
-        assert result.scalar_one() == metadata
+        assert 'https://example.test/new' in result.scalar_one()
+        async with engine.connect() as db:
+            assert 'langbot-resource://' in await db.scalar(sa.text('SELECT metadata_json FROM event_log'))
     finally:
         await engine.dispose()
 

tests/unit_tests/provider/test_request_policy.py

@@ -5,7 +5,10 @@ from unittest.mock import AsyncMock, Mock, patch
 import httpx
 import litellm
 import pytest
+from sqlalchemy.ext.asyncio import create_async_engine
 
+from langbot.pkg.entity.persistence.resource_url import ResourceURL
+from langbot.pkg.persistence import resource_urls
 from langbot.pkg.provider.modelmgr.errors import RequesterError
 from langbot.pkg.provider.modelmgr.request_policy import (
     CallState,
@@ -20,10 +23,22 @@ from tests.unit_tests.provider.conftest import TEST_EXECUTION_CONTEXT
 from tests.unit_tests.provider.test_litellmchat import MockRuntimeModel
 
 
+@pytest.fixture
+async def image_url_engine(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "resource_urls.db"}')
+    try:
+        async with engine.begin() as db:
+            await db.run_sync(ResourceURL.__table__.create)
+        resource_urls.install(engine.sync_engine)
+        yield engine
+    finally:
+        await engine.dispose()
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize('streaming', [False, True])
 async def test_image_credentials_are_checked_again_before_retry(
-    runtime_provider, runtime_llm_model, monkeypatch, streaming
+    runtime_provider, runtime_llm_model, monkeypatch, streaming, image_url_engine
 ):
     from langbot.pkg.platform.adapters.aiocqhttp.image_urls import OneBotImageURLs
     from langbot_plugin.api.entities.builtin.pipeline.query import Query
@@ -46,6 +61,7 @@ async def test_image_credentials_are_checked_again_before_retry(
         adapter=SimpleNamespace(bot=SimpleNamespace(connection=peer)),
     )
     runtime_provider.requester.ap.platform_mgr = SimpleNamespace(resolve_public_bot=AsyncMock(return_value=bot))
+    runtime_provider.requester.ap.persistence_mgr.get_db_engine = lambda: image_url_engine
     query = Query.model_construct(query_id=1, variables={}, bot_uuid='bot', pipeline_uuid='p', session=None)
     object.__setattr__(query, '_execution_context', TEST_EXECUTION_CONTEXT)
     original_url = 'https://multimedia.nt.qq.com.cn/download?appid=1407&fileid=file&rkey=old'