feat(storage): global trash cleanup, reference lists and browser profile status (C22–C24)

apps/server-python/src/openbot_server/attachment_references.py

@@ -1,7 +1,12 @@
-"""Owner-only metadata projection over retained channel Message and task identities."""
+"""Owner-only bounded counts and previews over retained channel Message and task identities."""
+from datetime import datetime
+
 from .control_errors import ControlError
+from .models import iso_timestamp
 
 REFERENCE_LIMIT = 10000
+LIST_LIMIT = 100
+_MARKER_PATTERN = r'\[openbot attachment: [0-9a-f]{8}(-[0-9a-f]{4}){3}-[0-9a-f]{12}\]'
 
 # Counts come from one SQL snapshot. Match the same canonical marker as task_store.attachment_ids;
 # existence counts a record once even if its text repeats the marker. No content leaves PostgreSQL.
@@ -37,3 +42,48 @@ async def with_reference_counts(db, channel_id, attachments):
     if len(counts) != len(attachments):
         raise ControlError(503, 'attachment_references_unavailable')
     return [{**item, 'referenceCount': counts[item['id']]} for item in attachments]
+
+
+_LIST_QUERY = """
+WITH message_refs AS MATERIALIZED (
+ SELECT id,created_at FROM messages
+ WHERE channel_id=%s AND position(%s in lower(content COLLATE "C"))>0 LIMIT %s
+), task_refs AS MATERIALIZED (
+ SELECT id,created_at FROM runs
+ WHERE channel_id=%s AND position(%s in lower(instruction COLLATE "C"))>0 LIMIT %s
+), message_page AS (
+ SELECT * FROM message_refs ORDER BY created_at DESC,id COLLATE "C" DESC LIMIT %s
+), task_page AS (
+ SELECT * FROM task_refs ORDER BY created_at DESC,id COLLATE "C" DESC LIMIT %s
+)
+SELECT (SELECT count(*) FROM message_refs) AS message_count,
+ (SELECT count(*) FROM task_refs) AS task_count,
+ COALESCE((SELECT jsonb_agg(jsonb_build_object('id',m.id,'createdAt',m.created_at,
+   'authorType',m.author_type,'authorId',m.author_id,
+   'preview',left(btrim(regexp_replace(m.content COLLATE "C",%s,'','gi')),120))
+   ORDER BY m.created_at DESC,m.id COLLATE "C" DESC)
+   FROM message_page p JOIN messages m ON m.id=p.id),'[]'::jsonb) AS messages,
+ COALESCE((SELECT jsonb_agg(jsonb_build_object('runId',r.id,'title',left(r.title,160),
+   'status',r.status,'createdAt',r.created_at) ORDER BY r.created_at DESC,r.id COLLATE "C" DESC)
+   FROM task_page p JOIN runs r ON r.id=p.id),'[]'::jsonb) AS tasks
+"""
+
+
+async def reference_list(db, channel_id, attachment_id, limit):
+    # Counts and both bounded pages share one statement/snapshot; Work mappings add no identity.
+    marker = '[openbot attachment: ' + attachment_id.lower() + ']'
+    row = await (await db.execute(_LIST_QUERY, (channel_id, marker, REFERENCE_LIMIT + 1,
+        channel_id, marker, REFERENCE_LIMIT + 1, limit, limit, _MARKER_PATTERN))).fetchone()
+    if row is None: raise ControlError(503, 'attachment_references_unavailable')
+    if any(type(row[key]) is not int or not 0 <= row[key] <= REFERENCE_LIMIT for key in ('message_count', 'task_count')):
+        raise ControlError(503, 'attachment_reference_limit')
+    messages = []
+    for item in row['messages']:
+        kind = {'human': 'owner', 'bot': 'bot', 'system': 'system'}[item['authorType']]
+        author = dict(kind=kind)
+        if kind == 'bot' and item['authorId'] is not None: author['botId'] = item['authorId']
+        messages.append(dict(id=item['id'], createdAt=iso_timestamp(datetime.fromisoformat(item['createdAt'])),
+            author=author, preview=item['preview']))
+    tasks = [{**item, 'createdAt': iso_timestamp(datetime.fromisoformat(item['createdAt']))} for item in row['tasks']]
+    return dict(messages=messages, tasks=tasks, messageCount=row['message_count'], taskCount=row['task_count'],
+        hasMore=row['message_count'] > len(messages) or row['task_count'] > len(tasks))

apps/server-python/src/openbot_server/browser_protocol.py

@@ -169,6 +169,7 @@ class Maintenance(Strict):
 
 class RuntimeState(Strict):
     running: bool
+    profileBytes: Annotated[int, BeforeValidator(integer), Field(ge=0, le=9007199254740991)] | None = None
 
 
 class MaintenanceInput(Strict):

apps/server-python/src/openbot_server/product_attachment_routes.py

@@ -13,7 +13,7 @@
 
 from fastapi.responses import Response
 
-from .attachment_references import with_reference_counts
+from .attachment_references import LIST_LIMIT, reference_list, with_reference_counts
 from .control_errors import ControlError
 from .http_input import read_attachment_upload, request_signal
 from .owner_files import MAX_BYTES
@@ -94,6 +94,18 @@ async def metadata(token,path,*_):
             return {'attachment':product.files.metadata(path['channel_id'],path['attachment_id'])}
     route(CHANNEL_BASE+'/{attachment_id}','GET',metadata)
 
+    async def references(token,path,_body,request):
+        async with product.files.lock(),product.transactions.transaction(token) as db:
+            await product.channel(db,path['channel_id'])
+            item = product.files.metadata(path['channel_id'],path['attachment_id'])
+            query = list(request.query_params.multi_items())
+            limit = request.query_params.get('limit', '20')
+            if (len(query) > 1 or any(key != 'limit' for key, _ in query) or not limit.isascii()
+                    or not limit.isdecimal() or len(limit) > 3 or not 1 <= int(limit) <= LIST_LIMIT):
+                raise ControlError(422, 'invalid_attachment_reference_query')
+            return await reference_list(db,path['channel_id'],item['id'],int(limit))
+    route(CHANNEL_BASE+'/{attachment_id}/references','GET',references)
+
     async def content(token,path,*_):
         async with product.files.lock(),product.transactions.transaction(token) as db:
             await product.channel(db,path['channel_id'])

apps/server-python/src/openbot_server/storage_routes.py

@@ -8,6 +8,11 @@ async def usage(token, _path, _body, request):
         return await product.storage.usage(token)
     route('/api/v1/storage', 'GET', usage)
 
+    async def cleanup_all(token, _path, body, request):
+        if request.query_params: raise ControlError(422, 'invalid_storage_query')
+        return await product.storage.cleanup_all(token, body)
+    route('/api/v1/storage/trash/cleanup', 'POST', cleanup_all, limit=4096)
+
     async def settings(token, *_): return await product.storage.settings(token)
     async def save(token, _path, body, _request): return await product.storage.save_settings(token, body)
     route('/api/v1/settings/storage', 'GET', settings)

apps/server-python/src/openbot_server/storage_service.py

@@ -172,6 +172,31 @@ async def settings(self, token):
         async with self.transactions.transaction(token) as db:
             return settings_projection(await (await db.execute("SELECT * FROM owner_storage_settings WHERE owner_id='owner'")).fetchone())
 
+    async def cleanup_all(self, token, command):
+        async with self.files.lock():
+            try:
+                async with self.transactions.transaction(token) as db:
+                    if (type(command) is not dict or set(command) != {'requestKey'} or type(command['requestKey']) is not str
+                            or not UUID.fullmatch(command['requestKey'])): raise ControlError(422, 'invalid_cleanup_request')
+                    key = command['requestKey'].lower()
+                    prior = await (await db.execute('SELECT response FROM storage_cleanup_receipts WHERE request_key=%s', (key,))).fetchone()
+                    if prior: result = prior['response']
+                    else:
+                        items, _ = self.catalog()
+                        trash = [item for item in items if item.get('deletedAt')]
+                        active = await (await db.execute('SELECT id FROM channels WHERE id=ANY(%s) AND deleted_at IS NULL ORDER BY id FOR SHARE',
+                            (sorted({item['channelId'] for item in trash}),))).fetchall()
+                        allowed = {row['id'] for row in active}
+                        result = await self._remove(db, [item for item in trash if item['channelId'] in allowed],
+                            actor='owner', reason='empty_trash_all')
+                        result['channelCount'] = len(allowed)
+                        await db.execute('INSERT INTO storage_cleanup_receipts(request_key,response) VALUES(%s,%s)', (key, Jsonb(result)))
+                await self.recover_locked()
+                return result
+            except BaseException:
+                await self.recover_locked()
+                raise
+
     async def save_settings(self, token, command):
         async with self.transactions.transaction(token) as db:
             if (type(command) is not dict or set(command) != {'expectedRevision', 'trashAutoPurgeDays'}

apps/server-python/src/openbot_server/storage_usage.py

@@ -45,6 +45,8 @@ async def measured_usage(db, files, object_root, artifact_root, channel, owner,
     attachment_prefix = files.root.relative_to(object_root).parts
     claimed = set()
     channel_active = channel_trash = owner_bytes = 0
+    referenced_ids = {item['id'] for item in trash if any(item['referenceCount'].values())}
+    referenced_bytes = 0
     channels = defaultdict(lambda: dict(sizeBytes=0, fileCount=0))
     for item in [*channel, *owner]:
         total = 0
@@ -60,6 +62,7 @@ async def measured_usage(db, files, object_root, artifact_root, channel, owner,
             channels[item['channelId']]['fileCount'] += 1
             if item.get('deletedAt'): channel_trash += total
             else: channel_active += total
+            if item['id'] in referenced_ids: referenced_bytes += total
     task_bytes = task_count = 0
     for key, size in objects.items():
         if key[0] == 'runs':
@@ -102,6 +105,6 @@ async def measured_usage(db, files, object_root, artifact_root, channel, owner,
             other=dict(sizeBytes=other, fileCount=sum(key not in claimed for key in objects)+extra_other_count),
             database=dict(sizeBytes=observed['bytes']), workingComputerBrowserData=None),
         trash=dict(fileCount=len(trash), sizeBytes=channel_trash,
-            referencedFileCount=sum(any(item['referenceCount'].values()) for item in trash)),
+            referencedFileCount=len(referenced_ids), referencedSizeBytes=referenced_bytes),
         topChannels=[dict(id=identity, name=names[identity]['name'], deleted=names[identity]['deleted_at'] is not None,
             **channels[identity]) for identity in top], topChannelsLimit=TOP_CHANNEL_LIMIT)

apps/server-python/tests/test_attachment_references.py

@@ -1,4 +1,4 @@
-"""Real disposable SQL/HTTP counts expose metadata, never the referencing content."""
+"""Real disposable SQL/HTTP checks for scoped counts and bounded Owner reference previews."""
 from uuid import uuid4
 
 import psycopg
@@ -21,6 +21,7 @@ def seed(synthetic_db):
     value={**synthetic_db,"bot":bot,"peer":peer,"channel":channel}
     yield value
     with psycopg.connect(synthetic_db["dsn"]) as db:
+        db.execute("DELETE FROM run_events WHERE channel_id=%s",(channel,))
         db.execute("DELETE FROM work_sources WHERE channel_id=%s",(channel,))
         db.execute("DELETE FROM work_tasks WHERE bot_id=ANY(%s)",([bot,peer],))
         db.execute("DELETE FROM runs WHERE channel_id=%s",(channel,))
@@ -55,7 +56,8 @@ def setup(seed, tmp_path):
         db.execute("INSERT INTO work_tasks(id,owner_id,bot_id,request_key,request_digest,objective,token_limit) VALUES (%s,'owner',%s,%s,%s,%s,0)",
             (task,seed['bot'],task,'a'*64,marker))
         db.execute('INSERT INTO work_sources(task_id,legacy_run_id,channel_id,source_message_id) VALUES (%s,%s,%s,%s)',(task,run1,seed['channel'],first))
-    app=create_app(PostgresReadStore(seed['dsn']),owner_name='Owner',secure_cookies=False,product=product)
+    app=create_app(PostgresReadStore(seed['dsn']),owner_name='Owner',secure_cookies=False,
+        allowed_origins=('http://testserver',),product=product)
     return product,item,unused,other,TestClient(app)
 
 
@@ -106,3 +108,81 @@ async def revoke_after_read(db,channel,items):
         api.cookies.set('openbot_session',seed['token'])
         response=api.get(f"/api/v1/channels/{seed['channel']}/attachments")
         assert response.status_code==401 and 'attachments' not in response.json()
+
+
+def test_reference_lists_keep_c19_matching_order_identity_and_short_previews(seed,tmp_path):
+    product,item,unused,other,api=setup(seed,tmp_path)
+    route=f"/api/v1/channels/{seed['channel']}/attachments/{item['id']}/references"
+    bot_message,system_message = str(uuid4()),str(uuid4())
+    marker=f"[OpenBot attachment: {item['id']}]"
+    with psycopg.connect(seed['dsn']) as db:
+        db.execute("INSERT INTO messages(id,channel_id,author_type,author_id,content) VALUES(%s,%s,'bot',%s,%s),(%s,%s,'system',NULL,%s)",
+            (bot_message,seed['channel'],seed['bot'],marker+'😀'*130+f"[OPENBOT ATTACHMENT: {unused['id']}]",
+             system_message,seed['channel'],'Scheduled '+marker.upper()))
+        db.execute("UPDATE messages SET created_at='2026-10-03T00:00:00Z' WHERE channel_id=%s",(seed['channel'],))
+        db.execute("UPDATE runs SET created_at='2026-10-03T00:00:00Z' WHERE channel_id=%s",(seed['channel'],))
+    with api:
+        assert api.get(route).status_code==401
+        api.cookies.set('openbot_session',seed['token'])
+        response=api.get(route)
+        assert response.status_code==200,response.text
+        result=response.json()
+        assert set(result)=={'messages','tasks','messageCount','taskCount','hasMore'}
+        assert result['messageCount']==4 and result['taskCount']==3 and result['hasMore'] is False
+        assert [m['id'] for m in result['messages']]==sorted((m['id'] for m in result['messages']),reverse=True)
+        assert [r['runId'] for r in result['tasks']]==sorted((r['runId'] for r in result['tasks']),reverse=True)
+        by_id={m['id']:m for m in result['messages']}
+        assert by_id[bot_message]['author']==dict(kind='bot',botId=seed['bot'])
+        assert by_id[bot_message]['preview']=='😀'*120
+        assert by_id[system_message]['author']==dict(kind='system')
+        assert by_id[system_message]['preview']=='Scheduled'
+        for value in result['messages']:
+            assert set(value)=={'id','createdAt','author','preview'}
+            assert len(value['preview'])<=120 and '[openbot attachment:' not in value['preview'].lower()
+        for value in result['tasks']: assert set(value)=={'runId','title','status','createdAt'}
+        assert {value['status'] for value in result['tasks']}=={'queued','completed'}
+        limited=api.get(route+'?limit=1').json()
+        assert limited=={**result,'messages':result['messages'][:1],'tasks':result['tasks'][:1],'hasMore':True}
+        product.files.set_deleted(seed['channel'],item['id'],True)
+        assert api.get(route).json()==result
+        assert api.get(route.replace(seed['channel'],other)).status_code==404
+
+
+def test_reference_list_query_limits_gone_files_and_unknown_counts_fail_closed(seed,tmp_path,monkeypatch):
+    from openbot_server import attachment_references
+    product,item,unused,_,api=setup(seed,tmp_path)
+    base=f"/api/v1/channels/{seed['channel']}/attachments/"
+    route=base+item['id']+'/references'
+    with api:
+        api.cookies.set('openbot_session',seed['token'])
+        for query in ('limit=0','limit=101','limit=-1','limit=1.5','limit=','limit=20&limit=20','offset=1','limit=２０'):
+            response=api.get(route+'?'+query)
+            assert response.status_code==422,response.text
+            assert 'messages' not in response.json()
+        assert api.get(route+'?limit=100').status_code==200
+        assert api.get(base+str(uuid4())+'/references').status_code==404
+        assert api.get(base+unused['id']+'/references').json()==dict(messages=[],tasks=[],messageCount=0,taskCount=0,hasMore=False)
+        product.files.set_deleted(seed['channel'],unused['id'],True)
+        assert api.delete(base+unused['id']+'/purge',headers={'Origin':'http://testserver'}).status_code==200
+        gone=api.get(base+unused['id']+'/references')
+        assert gone.status_code==410 and 'messages' not in gone.json()
+        monkeypatch.setattr(attachment_references,'REFERENCE_LIMIT',1)
+        refused=api.get(route)
+        assert refused.status_code==503 and refused.json()=={'error':'attachment_reference_limit'}
+
+
+def test_reference_list_rechecks_owner_before_releasing_content(seed,tmp_path,monkeypatch):
+    import hashlib
+    from openbot_server import product_attachment_routes
+    _,item,_,_,api=setup(seed,tmp_path)
+    original=product_attachment_routes.reference_list
+    async def revoke_after_read(db,*args):
+        result=await original(db,*args)
+        await db.execute('UPDATE auth_sessions SET revoked_at=now() WHERE token_digest=%s',
+            (hashlib.sha256(seed['token'].encode()).hexdigest(),))
+        return result
+    monkeypatch.setattr(product_attachment_routes,'reference_list',revoke_after_read)
+    with api:
+        api.cookies.set('openbot_session',seed['token'])
+        response=api.get(f"/api/v1/channels/{seed['channel']}/attachments/{item['id']}/references")
+        assert response.status_code==401 and 'PRIVATE' not in response.text and 'messages' not in response.json()