From 345666849743352286fce55704aea3426c3e4ffe Mon Sep 17 00:00:00 2001 From: BiFangKNT <1320414964@qq.com> Date: Fri, 23 Oct 2026 12:49:56 +0800 Subject: [PATCH] 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 | 3 +- skills/skills/langbot-mcp-ops/SKILL.md | 5 +- .../entity/persistence/agent_interaction.py | 7 +- .../pkg/entity/persistence/assistant.py | 3 +- .../pkg/entity/persistence/monitoring.py | 8 +- .../pkg/entity/persistence/pipeline.py | 3 +- .../pkg/entity/persistence/resource_url.py | 26 ++ .../alembic/versions/0041_resource_urls.py | 92 +++++ src/langbot/pkg/persistence/media.py | 48 ++- src/langbot/pkg/persistence/mgr.py | 9 + src/langbot/pkg/persistence/resource_urls.py | 391 ++++++++++++++++++ src/langbot/pkg/persistence/tenant_uow.py | 18 +- .../platform/adapters/aiocqhttp/image_urls.py | 51 ++- .../platform/adapters/aiocqhttp/transport.py | 1 + src/langbot/pkg/platform/image_urls.py | 11 +- .../persistence/test_monitoring_postgres.py | 39 ++ .../api/service/test_monitoring_executions.py | 12 +- .../persistence/test_resource_urls.py | 367 ++++++++++++++++ .../platform/test_onebot_image_urls.py | 51 ++- .../provider/test_request_policy.py | 18 +- 20 files changed, 1116 insertions(+), 47 deletions(-) create mode 100644 src/langbot/pkg/entity/persistence/resource_url.py create mode 100644 src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py create mode 100644 src/langbot/pkg/persistence/resource_urls.py create mode 100644 tests/unit_tests/persistence/test_resource_urls.py diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index 0cfee2568..a18f99565 100644 --- a/ARCHITECTURE.md +++ b/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. diff --git a/skills/skills/langbot-mcp-ops/SKILL.md b/skills/skills/langbot-mcp-ops/SKILL.md index 2af15c2f3..dde430706 100644 --- a/skills/skills/langbot-mcp-ops/SKILL.md +++ b/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. diff --git a/src/langbot/pkg/entity/persistence/agent_interaction.py b/src/langbot/pkg/entity/persistence/agent_interaction.py index c698f3b80..349f0be5a 100644 --- a/src/langbot/pkg/entity/persistence/agent_interaction.py +++ b/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) diff --git a/src/langbot/pkg/entity/persistence/assistant.py b/src/langbot/pkg/entity/persistence/assistant.py index 3a6c6c502..14b7d590a 100644 --- a/src/langbot/pkg/entity/persistence/assistant.py +++ b/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) diff --git a/src/langbot/pkg/entity/persistence/monitoring.py b/src/langbot/pkg/entity/persistence/monitoring.py index 6392363da..719823adc 100644 --- a/src/langbot/pkg/entity/persistence/monitoring.py +++ b/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__ = ( diff --git a/src/langbot/pkg/entity/persistence/pipeline.py b/src/langbot/pkg/entity/persistence/pipeline.py index 6451ed6b7..bd979493a 100644 --- a/src/langbot/pkg/entity/persistence/pipeline.py +++ b/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__ = ( diff --git a/src/langbot/pkg/entity/persistence/resource_url.py b/src/langbot/pkg/entity/persistence/resource_url.py new file mode 100644 index 000000000..6d9a19ced --- /dev/null +++ b/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'), + ) diff --git a/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py b/src/langbot/pkg/persistence/alembic/versions/0041_resource_urls.py new file mode 100644 index 000000000..0030a73b8 --- /dev/null +++ b/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) diff --git a/src/langbot/pkg/persistence/media.py b/src/langbot/pkg/persistence/media.py index 337c63834..817aee2ab 100644 --- a/src/langbot/pkg/persistence/media.py +++ b/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 diff --git a/src/langbot/pkg/persistence/mgr.py b/src/langbot/pkg/persistence/mgr.py index 9350f5f62..462a7e66e 100644 --- a/src/langbot/pkg/persistence/mgr.py +++ b/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() diff --git a/src/langbot/pkg/persistence/resource_urls.py b/src/langbot/pkg/persistence/resource_urls.py new file mode 100644 index 000000000..bdbde85b3 --- /dev/null +++ b/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) diff --git a/src/langbot/pkg/persistence/tenant_uow.py b/src/langbot/pkg/persistence/tenant_uow.py index 7de4cdba8..2b6a1093d 100644 --- a/src/langbot/pkg/persistence/tenant_uow.py +++ b/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 ""!r}' ) diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py b/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py index 0a8790189..701dafec6 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/image_urls.py +++ b/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 diff --git a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py b/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py index b1ac06ee9..63df03dfc 100644 --- a/src/langbot/pkg/platform/adapters/aiocqhttp/transport.py +++ b/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')) diff --git a/src/langbot/pkg/platform/image_urls.py b/src/langbot/pkg/platform/image_urls.py index fb6ef7001..6c16993f9 100644 --- a/src/langbot/pkg/platform/image_urls.py +++ b/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 diff --git a/tests/integration/persistence/test_monitoring_postgres.py b/tests/integration/persistence/test_monitoring_postgres.py index d0bfa6b41..b1c9931b0 100644 --- a/tests/integration/persistence/test_monitoring_postgres.py +++ b/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 diff --git a/tests/unit_tests/api/service/test_monitoring_executions.py b/tests/unit_tests/api/service/test_monitoring_executions.py index 2029c1e06..45b28ba55 100644 --- a/tests/unit_tests/api/service/test_monitoring_executions.py +++ b/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(): diff --git a/tests/unit_tests/persistence/test_resource_urls.py b/tests/unit_tests/persistence/test_resource_urls.py new file mode 100644 index 000000000..320baefd6 --- /dev/null +++ b/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')) diff --git a/tests/unit_tests/platform/test_onebot_image_urls.py b/tests/unit_tests/platform/test_onebot_image_urls.py index c11c934af..dd5afe60c 100644 --- a/tests/unit_tests/platform/test_onebot_image_urls.py +++ b/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() diff --git a/tests/unit_tests/provider/test_request_policy.py b/tests/unit_tests/provider/test_request_policy.py index 8161a5509..23143da0f 100644 --- a/tests/unit_tests/provider/test_request_policy.py +++ b/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'