From 9bcb357dd2d0c7cb367a73964e65550cd46de931 Mon Sep 17 00:00:00 2001
From: LIghtJUNction <lightjunction.me@gmail.com>
Date: Wed, 30 Sep 2026 23:41:58 +0800
Subject: [PATCH 1/4] feat: show public site activity and rules on homepage

---
 src/msg/plugins/discovery.py      |  52 ++++++++++++
 src/msg/transports/http_routes.py |  48 ++++++++++-
 tests/test_branding.py            | 131 +++++++++++++++++++++++++++++-
 3 files changed, 225 insertions(+), 6 deletions(-)

diff --git a/src/msg/plugins/discovery.py b/src/msg/plugins/discovery.py
index 89156335..ad5f4258 100644
--- a/src/msg/plugins/discovery.py
+++ b/src/msg/plugins/discovery.py
@@ -9,6 +9,7 @@
 from collections.abc import Mapping
 from dataclasses import replace as replace
 from datetime import timedelta
+from zoneinfo import ZoneInfo
 
 from msg.constants import (
     ADMINS_GROUP as ADMINS_GROUP,
@@ -569,6 +570,57 @@ def install(app):
     op, finish = registration(app, 'discovery', ('identity', 'content'))
     fields = {'type': 'array', 'items': STRING, 'maxItems': 30, 'uniqueItems': True}
 
+    @op('discovery.home', obj({}), effect='read')
+    async def home(ctx, request, tx):
+        """Live counts and recent resources, filtered by current read access."""
+        timezone = ZoneInfo('Asia/Taipei')
+        today = ctx.now.astimezone(timezone).date()
+        posts = posts_today = users = 0
+        latest = []
+        position = None
+        while True:
+            require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded')
+            seek = ' AND (r.created_at,r.id)<(?,?)' if position is not None else ''
+            rows = tx.rows(
+                'SELECT r.body,r.created_at,r.id FROM resources r '
+                "WHERE r.state='active' AND r.created_at<=? AND "
+                "(r.type='post' OR (r.type='user' AND EXISTS "
+                "(SELECT 1 FROM identities i WHERE i.id=r.id AND i.kind='subject')))"
+                + seek
+                + ' ORDER BY r.created_at DESC,r.id DESC LIMIT 128',
+                (wire(ctx.now), *(position if position is not None else ())),
+            )
+            for raw, created_at, rid in rows:
+                require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded')
+                position = (created_at, rid)
+                if not await visible(app, ctx, request, tx, rid):
+                    continue
+                resource = decode(Resource, loads(raw))
+                if resource.type == 'user':
+                    users += (await tx.subject(rid)).kind != 'system'
+                    continue
+                posts += 1
+                posts_today += resource.created_at.astimezone(timezone).date() == today
+                if len(latest) < 5:
+                    latest.append({
+                        'name': resource.name,
+                        'path': short_subject_path(await tx.path(rid)),
+                        'created_at': resource.created_at.astimezone(timezone).isoformat(),
+                    })
+            if len(rows) < 128:
+                break
+        require(time.monotonic() < ctx.deadline_monotonic, 'query_cost_exceeded')
+        return HandlerOutput(
+            data={
+                'posts': posts,
+                'posts_today': posts_today,
+                'users': users,
+                'date': today.isoformat(),
+                'timezone': timezone.key,
+                'latest': latest,
+            }
+        )
+
     @op(
         'discovery.get',
         obj(
diff --git a/src/msg/transports/http_routes.py b/src/msg/transports/http_routes.py
index 8c5c17cb..c2eaa515 100644
--- a/src/msg/transports/http_routes.py
+++ b/src/msg/transports/http_routes.py
@@ -71,8 +71,43 @@
 HOME_MARKDOWN = (
     '![msg.lmm.best logo](/favicon.png)\n\n# msg.lmm.best\n\n'
     'A place for people and agents to communicate clearly, share, and keep working.\n\n'
-    '[Agent guide](/AGENTS.md) · [Operations](/-/d)\n'
+    '[Topics](/main) · [Platform rules](/_rules) · [Agent guide](/AGENTS.md) · '
+    '[Operations](/-/d)\n'
 ).encode()
+
+
+def home_markdown(data=None):
+    lines = [
+        HOME_MARKDOWN.decode(),
+        '\n## Site activity\n',
+    ]
+    if data is None:
+        lines.append('Statistics and latest posts are temporarily unavailable.\n')
+    else:
+        lines.extend([
+            f'- Total public posts: {data["posts"]}',
+            f'- Posts today: {data["posts_today"]}',
+            f'- Public users: {data["users"]}',
+            f'\nToday: {data["date"]} ({data["timezone"]}). '
+            'Counts include active, publicly readable posts and user profiles; '
+            'system accounts are excluded.\n',
+            '## Latest posts\n',
+        ])
+        for item in data['latest']:
+            name = re.sub(r'([\\`*_{}\[\]<>!|&])', r'\\\1', item['name'])
+            lines.append(f'- [{name}]({quote(item["path"], safe="/@")}) — {item["created_at"]}')
+        if not data['latest']:
+            lines.append('No public posts yet.')
+    lines.extend([
+        '\n## Before posting\n',
+        'Read the [platform rules](/_rules) and [topic rules](/_rules/topics). '
+        'Public posts can be read by anyone. Ordinary resource links are read-only; '
+        'publishing or editing requires an authenticated operation. '
+        'The [community wiki](/wiki) provides guidance and cannot override platform rules.\n',
+    ])
+    return '\n'.join(lines).encode()
+
+
 HOME_HEADERS = {
     **BASE_HEADERS,
     'Content-Security-Policy': "default-src 'none'; style-src 'unsafe-inline'; img-src 'self'; base-uri 'none'; "
@@ -1748,15 +1783,22 @@ async def dispatch(request: Request):
                     headers=headers,
                 )
             if path == '/':
-                payload = HOME_MARKDOWN
+                packet = request_for(
+                    'discovery.home', {}, service.settings.service_url, source='manual'
+                )
+                result = await service.executor.execute(packet, entry='network')
+                if result.error and result.error.code != 'query_cost_exceeded':
+                    return json_response(result_wire(result), error_status(result.error.code))
+                payload = home_markdown(None if result.error else result.data)
                 headers = {
                     **BASE_HEADERS,
                     'Content-Length': str(len(payload)),
+                    'Cache-Control': 'no-store',
                 }
                 return Response(
                     b'' if request.method == 'HEAD' else payload,
                     media_type='text/plain'
-                    if 'text/html' in request.headers.get('accept', '')
+                    if 'text/html' in request.headers.get('accept', '').casefold()
                     else 'text/markdown',
                     headers=headers,
                 )
diff --git a/tests/test_branding.py b/tests/test_branding.py
index 1cc423ea..11bb58bb 100644
--- a/tests/test_branding.py
+++ b/tests/test_branding.py
@@ -1,11 +1,15 @@
 """The homepage is Markdown even for browsers; hosted files keep their own media type."""
 
 import re
+from dataclasses import replace
+from datetime import timedelta
 from importlib.resources import files
 
 import httpx
 import pytest
+from test_service import NOW, call, register
 
+from msg.core.codec import canonical, wire
 from msg.transports.http import create_app
 
 
@@ -27,9 +31,16 @@ async def test_logo_is_packaged_and_served_read_only(installed):
         assert agent.status_code == 200
         assert agent.headers['content-type'].startswith('text/markdown')
         assert '/AGENTS.md' in agent.text
+        assert '[Platform rules](/_rules)' in agent.text
+        assert 'Total public posts: 0' in agent.text
+        assert 'Posts today: 0' in agent.text
+        assert 'Public users: 0' in agent.text
+        assert 'No public posts yet.' in agent.text
+        assert 'Asia/Taipei' in agent.text
+        assert browser.headers['cache-control'] == 'no-store'
         assert 'sandbox' in agent.headers['content-security-policy']
         assert browser.content == agent.content
-        assert not re.search(r"[\u3400-\u9fff]", browser.text)
+        assert not re.search(r'[\u3400-\u9fff]', browser.text)
         browser_head = await http.head('/', headers={'Accept': 'text/html'})
         assert browser_head.content == b''
         assert browser_head.headers['content-type'].startswith('text/plain')
@@ -54,7 +65,7 @@ async def test_logo_is_packaged_and_served_read_only(installed):
         hosted = await http.get('/@root/web/index.html')
         assert hosted.status_code == 200 and b'<svg' in hosted.content
         assert '<html lang="en">' in hosted.text
-        assert not re.search(r"[\u3400-\u9fff]", hosted.text)
+        assert not re.search(r'[\u3400-\u9fff]', hosted.text)
         csp = hosted.headers['content-security-policy']
         assert "style-src 'unsafe-inline'" in csp
         assert 'font-src data:' in csp and 'img-src data:' in csp
@@ -69,13 +80,127 @@ async def test_logo_is_packaged_and_served_read_only(installed):
         )
 
 
+@pytest.mark.asyncio
+async def test_home_counts_dates_recent_posts_and_current_public_access(installed):
+    app, _ = installed
+    key, uid, cert = await register(app, 'home-author')
+    _, hidden_uid, _ = await register(app, 'hidden-profile')
+    async with app.metadata.transaction(write=True) as tx:
+        user = await tx.resource(hidden_uid)
+        await tx.replace(replace(user, mode=0o700, generation=user.generation + 1), user.generation)
+
+    # Taipei midnight is 16:00 UTC on the previous calendar day.
+    midnight = NOW.replace(hour=16) - timedelta(days=1)
+    cases = [
+        ('old', midnight - timedelta(microseconds=1), 0o644, 'active'),
+        ('at-midnight', midnight, 0o644, 'active'),
+        *[(f'recent-{i}', midnight + timedelta(minutes=i), 0o644, 'active') for i in range(1, 6)],
+        ('secret', NOW, 0o600, 'active'),
+        ('archived', NOW, 0o644, 'archived'),
+    ]
+    paths = {}
+    for name, created_at, mode, state in cases:
+        result = await call(
+            app,
+            'content.post_create',
+            {'parent': '/main', 'name': name, 'body': name},
+            key=key,
+            subject=uid,
+            certs=(cert,),
+        )
+        assert result.status == 'ok', result.error
+        async with app.metadata.transaction(write=True) as tx:
+            resource = await tx.resource(result.resources[0].id)
+            # Backdate fixture data directly; ordinary writes preserve creation facts.
+            resource = replace(resource, created_at=created_at)
+            tx.execute(
+                'UPDATE resources SET created_at=?,body=? WHERE id=?',
+                (wire(created_at), canonical(resource).decode(), resource.id),
+                write=True,
+            )
+            await tx.replace(
+                replace(
+                    resource,
+                    created_at=created_at,
+                    mode=mode,
+                    state=state,
+                    generation=resource.generation + 1,
+                ),
+                resource.generation,
+            )
+            paths[name] = await tx.path(resource.id)
+
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+    ) as http:
+        response = await http.get('/')
+        assert response.status_code == 200, response.text
+        assert 'Total public posts: 7' in response.text
+        assert 'Posts today: 6' in response.text
+        assert 'Public users: 1' in response.text
+        assert 'Today: 2026-09-27 (Asia/Taipei)' in response.text
+        latest = response.text.split('## Latest posts')[1].split('## Before posting')[0]
+        assert all(paths[f'recent-{i}'] in latest for i in range(1, 6))
+        assert paths['at-midnight'] not in latest and paths['old'] not in latest
+        assert paths['secret'] not in response.text and paths['archived'] not in response.text
+        assert latest.index('recent-5') < latest.index('recent-4') < latest.index('recent-1')
+        assert (await http.get(paths['recent-5'])).status_code == 200
+        head = await http.head('/')
+        assert head.content == b''
+        assert head.headers['content-length'] == str(len(response.content))
+
+        # Recompute on every visit; an ancestor becoming private hides its posts too.
+        async with app.metadata.transaction(write=True) as tx:
+            topic = await tx.resource(await tx.resolve('/main'))
+            await tx.replace(
+                replace(topic, mode=0o700, generation=topic.generation + 1), topic.generation
+            )
+        hidden = await http.get('/')
+        assert hidden.status_code == 200
+        assert 'Total public posts: 0' in hidden.text
+        assert 'Posts today: 0' in hidden.text
+        assert 'No public posts yet.' in hidden.text
+
+
+def test_home_escapes_post_names_and_paths():
+    from msg.transports.http_routes import home_markdown
+
+    page = home_markdown({
+        'posts': 1,
+        'posts_today': 1,
+        'users': 1,
+        'date': '2026-09-27',
+        'timezone': 'Asia/Taipei',
+        'latest': [
+            {'name': '[click]<script>&copy;', 'path': '/main/a (b).md', 'created_at': 'now'}
+        ],
+    }).decode()
+    assert r'\[click\]\<script\>' in page
+    assert r'\&copy;' in page
+    assert '(/main/a%20%28b%29.md)' in page
+
+
+def test_home_statistics_failure_keeps_rules_and_does_not_report_false_counts():
+    from msg.transports.http_routes import home_markdown
+
+    page = home_markdown().decode()
+    assert 'temporarily unavailable' in page
+    assert '[Platform rules](/_rules)' in page
+    assert '[topic rules](/_rules/topics)' in page
+    assert 'Total public posts: 0' not in page
+    assert 'No public posts yet.' not in page
+
+
 def test_game_script_permission_is_pinned_to_exact_bundled_root_page():
     from msg.bootstrap import ROOT_WEB_SAMPLE
     from msg.core.codec import digest
     from msg.extensions.hosting import HOSTED_HEADERS, hosted_headers
 
     released = digest(ROOT_WEB_SAMPLE)
-    assert 'allow-scripts' in hosted_headers('w_root_web', 'index.html', released)['Content-Security-Policy']
+    assert (
+        'allow-scripts'
+        in hosted_headers('w_root_web', 'index.html', released)['Content-Security-Policy']
+    )
     for site, path, content in (
         ('another_site', 'index.html', released),
         ('w_root_web', 'other.html', released),

From 32c5517f229fda7e373e2c4dd5c015cdb1a59fb3 Mon Sep 17 00:00:00 2001
From: LIghtJUNction <lightjunction.me@gmail.com>
Date: Wed, 30 Sep 2026 23:49:34 +0800
Subject: [PATCH 2/4] fix: preserve empty homepage HEAD errors and CI
 formatting

---
 src/msg/extensions/hosting.py     | 11 +++++++++--
 src/msg/transports/http_routes.py |  5 ++++-
 tests/test_ssh_bank_grant.py      | 16 ++++++++++++----
 3 files changed, 25 insertions(+), 7 deletions(-)

diff --git a/src/msg/extensions/hosting.py b/src/msg/extensions/hosting.py
index aa8ab8ab..98db6719 100644
--- a/src/msg/extensions/hosting.py
+++ b/src/msg/extensions/hosting.py
@@ -294,6 +294,7 @@ async def activate(ctx, request, tx):
     'Cache-Control': 'no-store',
 }
 
+
 def hosted_headers(site_id, file_path, blob_digest):
     """Only the exact bundled root introduction may run its own pinned script."""
     headers = dict(HOSTED_HEADERS)
@@ -312,12 +313,14 @@ def hosted_headers(site_id, file_path, blob_digest):
         for script in scripts
     )
     headers['Content-Security-Policy'] = (
-        "sandbox allow-scripts; default-src 'none'; script-src " + hashes
+        "sandbox allow-scripts; default-src 'none'; script-src "
+        + hashes
         + "; style-src 'unsafe-inline'; font-src data:; img-src data:; "
         "connect-src 'none'; base-uri 'none'; form-action 'none'; frame-ancestors 'none'"
     )
     return headers
 
+
 DOWNLOAD_TYPES = {
     'application/xhtml+xml',
     'image/svg+xml',
@@ -484,7 +487,11 @@ async def serve_hosted(service, request):
                     },
                 )
         etag = '"' + blob.digest + '"'
-        headers = {**hosted_headers(rid, file_path, blob.digest), 'ETag': etag, 'Accept-Ranges': 'bytes'}
+        headers = {
+            **hosted_headers(rid, file_path, blob.digest),
+            'ETag': etag,
+            'Accept-Ranges': 'bytes',
+        }
         media = blob.media_type.split(';', 1)[0].strip().lower()
         if media in DOWNLOAD_TYPES:
             headers['Content-Disposition'] = "attachment; filename*=UTF-8''" + quote(
diff --git a/src/msg/transports/http_routes.py b/src/msg/transports/http_routes.py
index c2eaa515..b253d69e 100644
--- a/src/msg/transports/http_routes.py
+++ b/src/msg/transports/http_routes.py
@@ -1788,7 +1788,10 @@ async def dispatch(request: Request):
                 )
                 result = await service.executor.execute(packet, entry='network')
                 if result.error and result.error.code != 'query_cost_exceeded':
-                    return json_response(result_wire(result), error_status(result.error.code))
+                    response = json_response(result_wire(result), error_status(result.error.code))
+                    if request.method == 'HEAD':
+                        response.body = b''
+                    return response
                 payload = home_markdown(None if result.error else result.data)
                 headers = {
                     **BASE_HEADERS,
diff --git a/tests/test_ssh_bank_grant.py b/tests/test_ssh_bank_grant.py
index 9d2a83bb..9e187a34 100644
--- a/tests/test_ssh_bank_grant.py
+++ b/tests/test_ssh_bank_grant.py
@@ -28,7 +28,9 @@ def test_bank_grant_flag_has_no_money_movement_or_revocation_path():
 
 @pytest.mark.parametrize('action', ['mint', 'burn', 'transfer', 'bank_fund', 'bank_remove'])
 def test_programmatic_opt_in_cannot_expand_to_other_root_actions(monkeypatch, action):
-    monkeypatch.setattr(money, 'require_ssh_administrator', lambda _: pytest.fail('SSH gate reached'))
+    monkeypatch.setattr(
+        money, 'require_ssh_administrator', lambda _: pytest.fail('SSH gate reached')
+    )
     with pytest.raises(Failure, match='ssh_bank_role_grant_only'):
         money.MoneyAdmin('/etc/msgd').execute(action, allow_ssh=True)
 
@@ -73,7 +75,9 @@ async def test_ssh_grant_still_requires_exact_confirmation_pin_and_root_signatur
     with pytest.raises(Failure, match='approval_cancelled'):
         await asyncio.to_thread(
             money.MoneyAdmin(app.settings.config_dir).execute,
-            'bank_add', subject_id=subject, allow_ssh=True,
+            'bank_add',
+            subject_id=subject,
+            allow_ssh=True,
         )
     assert not (await call(app, 'money.banks', {})).data['banks']
     seed, _, _, _ = installation_seed
@@ -87,13 +91,17 @@ async def test_ssh_grant_still_requires_exact_confirmation_pin_and_root_signatur
     with pytest.raises(Failure):
         await asyncio.to_thread(
             money.MoneyAdmin(app.settings.config_dir).execute,
-            'bank_add', subject_id=subject, allow_ssh=True,
+            'bank_add',
+            subject_id=subject,
+            allow_ssh=True,
         )
     assert not (await call(app, 'money.banks', {})).data['banks']
     monkeypatch.setattr(money.getpass, 'getpass', lambda _: 'correct-horse-test-passphrase')
     await asyncio.to_thread(
         money.MoneyAdmin(app.settings.config_dir).execute,
-        'bank_add', subject_id=subject, allow_ssh=True,
+        'bank_add',
+        subject_id=subject,
+        allow_ssh=True,
     )
     banks = (await call(app, 'money.banks', {})).data['banks']
     assert len(banks) == 1 and banks[0]['subject_id'] == subject and banks[0]['status'] == 'active'

From 278f6ef970eddb7adc1bfd2d35db3cee04d978c0 Mon Sep 17 00:00:00 2001
From: LIghtJUNction <lightjunction.me@gmail.com>
Date: Thu, 1 Oct 2026 00:24:53 +0800
Subject: [PATCH 3/4] test: adapt HTTP boundary fixtures to homepage discovery

---
 tests/test_nginx_secret_logs.py     | 32 ++++++++++++++++++++++++++---
 tests/test_request_target_safety.py | 29 +++++++++++++++++++++++++-
 2 files changed, 57 insertions(+), 4 deletions(-)

diff --git a/tests/test_nginx_secret_logs.py b/tests/test_nginx_secret_logs.py
index ea0e6aae..46a27377 100644
--- a/tests/test_nginx_secret_logs.py
+++ b/tests/test_nginx_secret_logs.py
@@ -1,7 +1,7 @@
 """Run the complete deployment config, including TLS/default/error contexts.
 
-The application fixture has no business executor: only actual HTTP boundary
-rejections and health/root reads are exercised here. PostgreSQL/authentication
+The application fixture only stubs read-only homepage discovery: actual HTTP
+boundary rejections and health/root reads are exercised here. PostgreSQL/authentication
 and zero-write assertions remain in test_url_secrets.py.
 """
 
@@ -147,14 +147,40 @@ def _proxy(tmp_path, backend_port, *, unsafe_control=False):
 
 @contextmanager
 def _application():
+    class HomepageExecutor:
+        async def require_current_runtime(self):
+            pass
+
+        def recovery_drill_active(self):
+            return False
+
+        async def execute(self, packet, *, entry):
+            # The homepage is the only read exercised by this boundary fixture.
+            # Any attempted business execution still fails immediately.
+            assert packet.operation == 'discovery.home'
+            assert entry == 'network'
+            assert not packet.arguments
+            return SimpleNamespace(
+                error=None,
+                data={
+                    'posts': 0,
+                    'posts_today': 0,
+                    'users': 0,
+                    'date': '2026-09-30',
+                    'timezone': 'Asia/Taipei',
+                    'latest': [],
+                },
+            )
+
     class BoundaryRegistry:
         def operation(self, name, *_args):
             # These specs only enable real packet parsing/rejection. Any request
-            # reaching business execution fails because there is no executor.
+            # reaching business execution fails in the homepage-only executor.
             return SimpleNamespace(name=name, entries=('network',), effect='transaction')
 
     service = SimpleNamespace(
         _loaded=True,
+        executor=HomepageExecutor(),
         registry=BoundaryRegistry(),
         settings=SimpleNamespace(
             service_url='https://msg.example.org',
diff --git a/tests/test_request_target_safety.py b/tests/test_request_target_safety.py
index 4d2765d6..c18e339e 100644
--- a/tests/test_request_target_safety.py
+++ b/tests/test_request_target_safety.py
@@ -1,4 +1,4 @@
-"""Raw URL boundaries without a database; no business execution is simulated."""
+"""Raw URL boundaries without a database; only homepage discovery is stubbed."""
 
 from types import SimpleNamespace
 from uuid import uuid4
@@ -16,8 +16,35 @@
 
 def boundary_service():
     """Only health/root and rejection paths are usable on this service."""
+
+    class HomepageExecutor:
+        async def require_current_runtime(self):
+            pass
+
+        def recovery_drill_active(self):
+            return False
+
+        async def execute(self, packet, *, entry):
+            # The homepage is the only read exercised by this boundary fixture.
+            # Any attempted business execution still fails immediately.
+            assert packet.operation == 'discovery.home'
+            assert entry == 'network'
+            assert not packet.arguments
+            return SimpleNamespace(
+                error=None,
+                data={
+                    'posts': 0,
+                    'posts_today': 0,
+                    'users': 0,
+                    'date': '2026-09-30',
+                    'timezone': 'Asia/Taipei',
+                    'latest': [],
+                },
+            )
+
     return SimpleNamespace(
         _loaded=True,
+        executor=HomepageExecutor(),
         settings=SimpleNamespace(
             service_url='http://testserver',
             server=SimpleNamespace(

From c283fa57f8e10cec80db7b8070ecf105e5ef94fb Mon Sep 17 00:00:00 2001
From: LIghtJUNction <lightjunction.me@gmail.com>
Date: Thu, 1 Oct 2026 00:33:51 +0800
Subject: [PATCH 4/4] Fix scoped read, content, authorization and delivery
 contract defects (#201)

* Reject non-integer published Registry contract versions

* Reject malformed legacy share sources without vetoing independent grants

Refs #76. Revalidate legacy creation and expiration times during reads, matching the live v2 boundary. PostgreSQL regression covers malformed and future facts, revision/history reads, independent v2 access, and final revocation.

* fix(hosting): reject unreachable deployment paths

* Reject unified patches that join EOF into untouched lines

* Fix exact search phrases crossing field boundaries

* Reject explicit zero ports in webhook endpoints

Refs #81. Preserve the HTTPS-only 443 endpoint policy for explicit :0 and reject it before network setup.

* Fix RSS limit validation for nondecimal Unicode digits

* test: cover honor ceremony concurrency and request replay

* Bind bounty escrow accounts to their listing before use

* Fix explicit zero-port network policy checks

* style: format integrated changes and current baseline

* Exercise Registry version defenses through plugin adapters
---
 src/msg/core/registry.py                    |  11 +-
 src/msg/core/text_patch.py                  |   9 ++
 src/msg/extensions/hosting.py               |   6 +
 src/msg/plugins/bounty.py                   |  10 ++
 src/msg/plugins/discovery.py                |   5 +-
 src/msg/security/authorization.py           |  19 +--
 src/msg/security/network.py                 |   4 +-
 src/msg/transports/http_routes.py           |   2 +-
 src/msg/workers/webhook.py                  |   2 +-
 tests/test_achievement_concurrency.py       | 125 ++++++++++++++++++++
 tests/test_bounty_escrow_binding.py         |  68 +++++++++++
 tests/test_hosting_addressable_paths.py     | 107 +++++++++++++++++
 tests/test_legacy_share_source_isolation.py |  53 +++++++++
 tests/test_network_url_ports.py             |  45 +++++++
 tests/test_registry_contract_references.py  |  56 ++++++++-
 tests/test_rss_query_validation.py          |  43 +++++++
 tests/test_search_phrase_boundaries.py      |  54 +++++++++
 tests/test_text_patch_eof_boundaries.py     | 117 ++++++++++++++++++
 tests/test_webhook_delivery.py              |  32 ++++-
 19 files changed, 752 insertions(+), 16 deletions(-)
 create mode 100644 tests/test_achievement_concurrency.py
 create mode 100644 tests/test_bounty_escrow_binding.py
 create mode 100644 tests/test_hosting_addressable_paths.py
 create mode 100644 tests/test_legacy_share_source_isolation.py
 create mode 100644 tests/test_network_url_ports.py
 create mode 100644 tests/test_rss_query_validation.py
 create mode 100644 tests/test_search_phrase_boundaries.py
 create mode 100644 tests/test_text_patch_eof_boundaries.py

diff --git a/src/msg/core/registry.py b/src/msg/core/registry.py
index 2fbaeb62..ce00a4c6 100644
--- a/src/msg/core/registry.py
+++ b/src/msg/core/registry.py
@@ -37,7 +37,10 @@ def __init__(self):
 
     def _insert(self, collection, spec, kind):
         require(not self._frozen, 'registry_frozen')
-        require(spec.version >= 1 and spec.name and '*' not in spec.name, 'invalid_registry_name')
+        require(
+            type(spec.version) is int and spec.version >= 1 and spec.name and '*' not in spec.name,
+            'invalid_registry_name',
+        )
         key = (spec.name, spec.version)
         require(key not in collection, 'duplicate_' + kind)
         collection[key] = spec
@@ -126,7 +129,11 @@ def add(self, manifest):
             )
         for item in (*manifest.resource_types, *manifest.capabilities, *manifest.operations):
             require(
-                item.version >= 1 and item.name and '*' not in item.name, 'invalid_registry_name'
+                type(item.version) is int
+                and item.version >= 1
+                and item.name
+                and '*' not in item.name,
+                'invalid_registry_name',
             )
         for item in manifest.operations:
             require(
diff --git a/src/msg/core/text_patch.py b/src/msg/core/text_patch.py
index 7dc31c43..9504333e 100644
--- a/src/msg/core/text_patch.py
+++ b/src/msg/core/text_patch.py
@@ -150,6 +150,9 @@ def _hunks(text):
                 previous = None
             else:
                 require(line[:1] in {' ', '-', '+'}, 'invalid_unified_patch')
+                # A missing file newline must be represented by the explicit
+                # marker, not an unterminated diff transport line.
+                require(line.endswith('\n'), 'invalid_unified_patch')
                 # Once counts are exhausted, another file header is not a hunk.
                 require(len(old) < old_count or len(new) < new_count, 'invalid_unified_patch')
                 previous = []
@@ -195,6 +198,12 @@ def _unified(source, hunks, *, rebase=False):
         require(last <= start <= len(lines), 'invalid_unified_patch')
         end = start + len(hunk.old)
         require(tuple(lines[start:end]) == hunk.old, 'patch_no_match')
+        # A new EOF marker cannot consume the separator before untouched
+        # source lines and silently concatenate their bytes into this line.
+        require(
+            not hunk.new or hunk.new[-1].endswith('\n') or end == len(lines),
+            'invalid_unified_patch',
+        )
         result.extend(lines[last:start])
         result.extend(hunk.new)
         last = end
diff --git a/src/msg/extensions/hosting.py b/src/msg/extensions/hosting.py
index 98db6719..ec5aee16 100644
--- a/src/msg/extensions/hosting.py
+++ b/src/msg/extensions/hosting.py
@@ -40,11 +40,17 @@ def path_name(value):
         and all(ord(c) >= 32 and ord(c) != 127 for c in value),
         'invalid_hosting_path',
     )
+    # Hosted reads deliberately reject percent-encoded and non-ASCII paths.
+    # Reject names requiring URL escaping before accepting a deployment that
+    # could never be reached through that boundary.
+    require(quote(value, safe="/:@!$&'()*+,;=") == value, 'invalid_hosting_path')
     parts = value.split('/')
     require(
         all(part not in {'', '.', '..'} and not part.startswith('.') for part in parts),
         'invalid_hosting_path',
     )
+    # These first components select preview/history routes rather than files.
+    require(len(parts) == 1 or parts[0] not in {'_preview', '_rev'}, 'invalid_hosting_path')
     return '/'.join(parts)
 
 
diff --git a/src/msg/plugins/bounty.py b/src/msg/plugins/bounty.py
index 9b511ce2..25111a9f 100644
--- a/src/msg/plugins/bounty.py
+++ b/src/msg/plugins/bounty.py
@@ -135,6 +135,16 @@ async def validate_contract(app, tx, row):
     """Mutable accounting state cannot change a publisher's immutable terms."""
     from msg.market.catalog import read_listing_body
 
+    # A mixed restore must not redirect payouts, refunds or top-ups to another
+    # listing's escrow or a signable Subject account.
+    account = tx.one(
+        'SELECT kind,subject_id,source_id FROM ledger_accounts WHERE id=?',
+        (row['escrow_subject'],),
+    )
+    require(
+        account == ('bounty_escrow', None, row['listing_id']),
+        'bounty_escrow_mismatch',
+    )
     resource = await tx.resource(row['listing_id'])
     body, _ = await read_listing_body(app, tx, resource)
     require(
diff --git a/src/msg/plugins/discovery.py b/src/msg/plugins/discovery.py
index ad5f4258..4148c278 100644
--- a/src/msg/plugins/discovery.py
+++ b/src/msg/plugins/discovery.py
@@ -1673,7 +1673,10 @@ async def endpoint_visible(rid):
                 else any(term in whole for term in terms)
             ):
                 continue
-            if exact and exact not in whole:
+            # Fields are separate documents for phrase matching. Joining them
+            # invents phrases at name/body/metadata boundaries that cannot be
+            # represented by a snippet or a matched field.
+            if exact and not any(exact in value for value in lowered.values()):
                 continue
             if any(term in whole for term in excluded):
                 continue
diff --git a/src/msg/security/authorization.py b/src/msg/security/authorization.py
index 9e27e30b..4a22c0d4 100644
--- a/src/msg/security/authorization.py
+++ b/src/msg/security/authorization.py
@@ -196,17 +196,20 @@ async def shared_read(self, principal, resource, chain, now, session):
         if share_target_error(self.registry, resource, chain, session):
             return False
         row = session.one(
-            """SELECT grantor,expires_at FROM share_grants
+            """SELECT grantor,expires_at,created_at FROM share_grants
             WHERE resource_id=? AND grantee=? AND revoked_at IS NULL""",
             (resource.id, principal.subject),
         )
-        if (
-            row
-            and row[0] == resource.owner
-            and parse_time(row[1]) > now
-            and resource.state == 'active'
-        ):
-            return True
+        if row and row[0] == resource.owner and resource.state == 'active':
+            # Restored legacy facts have the same time boundary as v2 sources.
+            # A malformed source denies only itself, preserving independent ones.
+            try:
+                expires, created = parse_time(row[1]), parse_time(row[2])
+            except Failure, ValueError, TypeError, OverflowError:
+                pass
+            else:
+                if created <= now < expires:
+                    return True
         if resource.state != 'active':
             return False
         candidates = session.rows(
diff --git a/src/msg/security/network.py b/src/msg/security/network.py
index f10d4c0b..03f43f84 100644
--- a/src/msg/security/network.py
+++ b/src/msg/security/network.py
@@ -59,7 +59,9 @@ def validate_url(url, method, policy):
             'invalid_network_url',
         )
         host = normalized_host(parsed.hostname)
-        port = parsed.port or (443 if parsed.scheme == 'https' else 80)
+        port = parsed.port
+        if port is None:
+            port = 443 if parsed.scheme == 'https' else 80
     except ValueError as exc:
         if isinstance(exc, Failure):
             raise
diff --git a/src/msg/transports/http_routes.py b/src/msg/transports/http_routes.py
index b253d69e..9d259374 100644
--- a/src/msg/transports/http_routes.py
+++ b/src/msg/transports/http_routes.py
@@ -821,7 +821,7 @@ async def dispatch(request: Request):
                 args = dict(request.query_params)
                 require(set(args) <= {'parent', 'limit', 'cursor'}, 'unknown_query_parameter')
                 if 'limit' in args:
-                    require(args['limit'].isdigit(), 'invalid_limit')
+                    require(args['limit'].isdecimal(), 'invalid_limit')
                     args['limit'] = int(args['limit'])
                 header = request.headers.get('x-msg-request')
                 packet = (
diff --git a/src/msg/workers/webhook.py b/src/msg/workers/webhook.py
index 2a331a1a..f63fbb98 100644
--- a/src/msg/workers/webhook.py
+++ b/src/msg/workers/webhook.py
@@ -56,7 +56,7 @@ def validate_endpoint(url: str) -> tuple[str, int]:
             'invalid_webhook_url',
         )
         host = normalized_host(parsed.hostname)
-        port = parsed.port or 443
+        port = 443 if parsed.port is None else parsed.port
     except ValueError as exc:
         raise Failure('invalid_webhook_url') from exc
     require(port == 443, 'webhook_port_forbidden')
diff --git a/tests/test_achievement_concurrency.py b/tests/test_achievement_concurrency.py
new file mode 100644
index 00000000..f5992435
--- /dev/null
+++ b/tests/test_achievement_concurrency.py
@@ -0,0 +1,125 @@
+"""Concurrent and retried ceremonies keep one challenge and one honor fact."""
+
+import asyncio
+
+import pytest
+from test_achievements import advance
+from test_service import call, register
+
+from msg.core.codec import loads, wire
+from msg.plugins.achievements import FINAL_STATEMENT
+
+
+@pytest.mark.asyncio
+async def test_concurrent_starts_keep_the_successful_challenge_usable(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-concurrent-start')
+    results = await asyncio.gather(
+        *(
+            call(app, 'achievement.start', {}, key=key, subject=subject, rid=f'start-{index}')
+            for index in range(2)
+        )
+    )
+    successes = [result for result in results if result.status == 'ok']
+    errors = [result for result in results if result.status == 'error']
+    assert len(successes) == len(errors) == 1, [wire(result) for result in results]
+    assert errors[0].error.code == 'achievement_ceremony_active'
+    async with app.metadata.transaction(write=False) as tx:
+        rows = tx.rows('SELECT id,state FROM achievement_ceremonies WHERE subject=?', (subject,))
+        assert rows == [(successes[0].data['challenge_id'], 'active')]
+    advanced = await advance(app, key, subject, successes[0].data)
+    assert advanced.status == 'ok' and advanced.data['round'] == 2, wire(advanced)
+
+
+@pytest.mark.asyncio
+async def test_answer_request_replay_preserves_nonce_without_advancing_again(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-answer-replay')
+    challenge = (await call(app, 'achievement.start', {}, key=key, subject=subject)).data
+    arguments = {
+        name: challenge[name] for name in ('challenge_id', 'round', 'question_digest', 'nonce')
+    }
+    arguments['answer'] = 'y'
+    results = await asyncio.gather(
+        *(
+            call(app, 'achievement.answer', arguments, key=key, subject=subject, rid='answer-once')
+            for _ in range(2)
+        )
+    )
+    assert all(result.status == 'ok' for result in results), [wire(result) for result in results]
+    assert sum(result.replayed for result in results) == 1
+    assert results[0].data == results[1].data
+    assert results[0].data['round'] == 2
+    assert results[0].data['nonce'] != challenge['nonce']
+    third = await advance(app, key, subject, results[0].data)
+    assert third.status == 'ok' and third.data['round'] == 3, wire(third)
+    replay = await call(
+        app, 'achievement.answer', arguments, key=key, subject=subject, rid='answer-once'
+    )
+    assert replay.status == 'ok' and replay.replayed
+    assert replay.data == results[0].data
+    async with app.metadata.transaction(write=False) as tx:
+        state = loads(
+            tx.one(
+                'SELECT body FROM achievement_ceremonies WHERE id=?', (challenge['challenge_id'],)
+            )[0]
+        )
+        assert state['status'] == 'active' and state['round'] == 3
+        assert [answer['round'] for answer in state['answers']] == [1, 2]
+
+
+@pytest.mark.asyncio
+async def test_concurrent_finish_and_replay_issue_only_one_grant_and_audit(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'honor-concurrent-finish')
+    challenge = (await call(app, 'achievement.start', {}, key=key, subject=subject)).data
+    for _ in range(4):
+        result = await advance(app, key, subject, challenge)
+        assert result.status == 'ok', wire(result)
+        challenge = result.data
+    arguments = {
+        name: challenge[name]
+        for name in ('challenge_id', 'question_digest', 'nonce', 'ceremony_digest')
+    }
+    arguments['statement'] = FINAL_STATEMENT
+    results = await asyncio.gather(
+        *(
+            call(
+                app,
+                'achievement.finish',
+                arguments,
+                key=key,
+                subject=subject,
+                rid=f'finish-{index}',
+            )
+            for index in range(2)
+        )
+    )
+    successful = [index for index, result in enumerate(results) if result.status == 'ok']
+    assert len(successful) == 1, [wire(result) for result in results]
+    winner = successful[0]
+    assert results[1 - winner].error.code == 'ceremony_inactive'
+    replay = await call(
+        app,
+        'achievement.finish',
+        arguments,
+        key=key,
+        subject=subject,
+        rid=f'finish-{winner}',
+    )
+    assert replay.status == 'ok' and replay.replayed
+    assert replay.data == results[winner].data
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM achievement_grants WHERE subject=?', (subject,))[0] == 1
+        audits = [loads(row[0]) for row in tx.rows('SELECT body FROM audit')]
+    issued = [
+        audit
+        for audit in audits
+        if audit['event']['type'] == 'achievement.auto.issue'
+        and audit['event']['subject'] == subject
+    ]
+    assert len(issued) == 1
+    assert issued[0]['event']['data']['certificate_id'] == replay.data['grant']['id']
+    listed = await call(app, 'achievement.list', {'subject_id': subject})
+    assert listed.status == 'ok' and len(listed.data['achievements']) == 1
+    assert listed.data['achievements'][0] == replay.data['grant']
diff --git a/tests/test_bounty_escrow_binding.py b/tests/test_bounty_escrow_binding.py
new file mode 100644
index 00000000..14da84a7
--- /dev/null
+++ b/tests/test_bounty_escrow_binding.py
@@ -0,0 +1,68 @@
+"""Mixed restored projections cannot spend an unrelated ledger account."""
+
+import pytest
+from test_bounty import fund, terms
+from test_market_72 import proof
+from test_service import call, register
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('target', ['subject', 'other_bounty'])
+@pytest.mark.parametrize(
+    'operation', ['bounty.claim', 'bounty.close', 'bounty.top_up', 'bounty.get']
+)
+async def test_bounty_escrow_must_belong_to_its_listing(installed, target, operation):
+    app, _ = installed
+    publisher_key, publisher, _ = await register(app, 'escrow-binding-publisher')
+    claimant_key, claimant, _ = await register(app, 'escrow-binding-claimant')
+    await fund(app, publisher, 40)
+    created = await call(app, 'bounty.create', terms(), key=publisher_key, subject=publisher)
+    assert created.status == 'ok', wire(created)
+    listing_id = created.data['bounty']['listing_id']
+    claim = await proof(app, (claimant_key, claimant), listing_id)
+    wrong_account = publisher
+    if target == 'other_bounty':
+        other = await call(
+            app, 'bounty.create', terms(name='other-escrow'), key=publisher_key, subject=publisher
+        )
+        assert other.status == 'ok', wire(other)
+        wrong_account = other.data['funding']['body']['to_subject']
+    async with app.metadata.transaction(write=True) as tx:
+        if target == 'other_bounty':
+            # Keep the unique escrow pointer constraint intact while emulating
+            # inconsistent materialized rows from different restored snapshots.
+            tx.execute(
+                'UPDATE bounty_listings SET escrow_subject=? WHERE listing_id=?',
+                (publisher, other.data['bounty']['listing_id']),
+                write=True,
+            )
+        tx.execute(
+            'UPDATE bounty_listings SET escrow_subject=? WHERE listing_id=?',
+            (wrong_account, listing_id),
+            write=True,
+        )
+        before_ledger = tx.rows('SELECT * FROM money_ledger ORDER BY id')
+        before_challenge = tx.one(
+            'SELECT * FROM bounty_challenges WHERE id=?', (claim['challenge_id'],)
+        )
+    if operation == 'bounty.claim':
+        params, key, subject = claim, claimant_key, claimant
+    else:
+        params, key, subject = {'listing_id': listing_id}, publisher_key, publisher
+        if operation == 'bounty.top_up':
+            params['amount_minor'] = 1
+    result = await call(app, operation, params, key=key, subject=subject)
+    assert result.status == 'error' and result.error.code == 'bounty_escrow_mismatch', wire(result)
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.rows('SELECT * FROM money_ledger ORDER BY id') == before_ledger
+        assert (
+            tx.one('SELECT * FROM bounty_challenges WHERE id=?', (claim['challenge_id'],))
+            == before_challenge
+        )
+        assert tx.one('SELECT COUNT(*) FROM bounty_claims')[0] == 0
+        assert (
+            tx.one("SELECT COUNT(*) FROM messages WHERE body::jsonb->>'source'='bounty_claim'")[0]
+            == 0
+        )
diff --git a/tests/test_hosting_addressable_paths.py b/tests/test_hosting_addressable_paths.py
new file mode 100644
index 00000000..183528d1
--- /dev/null
+++ b/tests/test_hosting_addressable_paths.py
@@ -0,0 +1,107 @@
+"""Deployments reject names the hosted HTTP boundary cannot address."""
+
+import httpx
+import pytest
+from test_service import call, register
+
+from msg.core.codec import b64, wire
+from msg.transports.http import create_app
+
+
+@pytest.mark.parametrize('operation', ['hosting.deploy', 'hosting.preview'])
+@pytest.mark.asyncio
+async def test_unaddressable_paths_are_rejected_before_materialization(installed, operation):
+    app, _ = installed
+    key, user, _ = await register(app, 'path-owner')
+    site = await call(
+        app, 'hosting.create', {'parent': '/@path-owner', 'name': 'web'}, key=key, subject=user
+    )
+    source = await call(
+        app,
+        'content.file_put',
+        {
+            'parent': '/@path-owner/files',
+            'name': 'page.html',
+            'data': b64(b'<h1>addressable</h1>'),
+            'media_type': 'text/html',
+        },
+        key=key,
+        subject=user,
+    )
+    assert site.status == source.status == 'ok'
+    async with app.metadata.transaction(write=False) as tx:
+        before = tuple(
+            tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+            for table in ('resources', 'revisions', 'settings')
+        )
+    for path in (
+        'hello world.html',
+        '中文.html',
+        'a%20b.html',
+        'a#b.html',
+        'a?b.html',
+        '_preview/fake/index.html',
+        '_rev/fake/index.html',
+    ):
+        result = await call(
+            app,
+            operation,
+            {
+                'id': site.resources[0].id,
+                'entries': [{'path': path, 'source': wire(source.resources[0])}],
+            },
+            key=key,
+            subject=user,
+            expected=((site.resources[0].id, site.data['generation']),),
+        )
+        assert result.status == 'error' and result.error.code == 'invalid_hosting_path', wire(
+            result
+        )
+        async with app.metadata.transaction(write=False) as tx:
+            assert (
+                tuple(
+                    tx.one(f'SELECT COUNT(*) FROM {table}')[0]
+                    for table in ('resources', 'revisions', 'settings')
+                )
+                == before
+            )
+
+
+@pytest.mark.parametrize('path', ["assets/v1/page-_.~!$&'()*+,;=:@.html", '_rev', '_preview'])
+@pytest.mark.asyncio
+async def test_uri_safe_paths_remain_reachable(installed, path):
+    app, _ = installed
+    key, user, _ = await register(app, 'safe-path-owner')
+    site = await call(
+        app, 'hosting.create', {'parent': '/@safe-path-owner', 'name': 'web'}, key=key, subject=user
+    )
+    html = b'<h1>nested path</h1>'
+    source = await call(
+        app,
+        'content.file_put',
+        {
+            'parent': '/@safe-path-owner/files',
+            'name': 'page.html',
+            'data': b64(html),
+            'media_type': 'text/html',
+        },
+        key=key,
+        subject=user,
+    )
+    deployed = await call(
+        app,
+        'hosting.deploy',
+        {
+            'id': site.resources[0].id,
+            'entries': [{'path': path, 'source': wire(source.resources[0])}],
+        },
+        key=key,
+        subject=user,
+        expected=((site.resources[0].id, site.data['generation']),),
+    )
+    assert deployed.status == 'ok', wire(deployed)
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=create_app(app)), base_url='http://testserver'
+    ) as http:
+        served = await http.get('/@safe-path-owner/web/' + path)
+    assert served.status_code == 200 and served.content == html
diff --git a/tests/test_legacy_share_source_isolation.py b/tests/test_legacy_share_source_isolation.py
new file mode 100644
index 00000000..26734197
--- /dev/null
+++ b/tests/test_legacy_share_source_isolation.py
@@ -0,0 +1,53 @@
+"""Restored legacy shares fail closed without vetoing an independent source."""
+
+from datetime import timedelta
+
+import pytest
+from test_authorization_sources import grant, private_post
+from test_service import NOW, register
+from test_share_grants_v2 import invoke, ok
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('column', 'value'),
+    [
+        ('expires_at', 'corrupt-restored-time'),
+        ('created_at', 'corrupt-restored-time'),
+        ('created_at', wire(NOW + timedelta(hours=1))),
+    ],
+)
+async def test_malformed_legacy_source_does_not_veto_independent_share(installed, column, value):
+    app, _ = installed
+    owner = await register(app, 'legacy-source-owner')
+    reader = await register(app, 'legacy-source-reader')
+    rid, revision = await private_post(app, owner)
+    legacy = ok(
+        await invoke(
+            app,
+            owner,
+            'sharing.grant',
+            {'resource': rid, 'grantee': reader[1], 'expires_at': wire(NOW + timedelta(days=1))},
+        )
+    ).data['grant']['id']
+    queries = ({'id': rid}, {'id': rid, 'view': 'history'}, {'id': rid, 'revision': revision})
+    for query in queries:
+        ok(await invoke(app, reader, 'discovery.get', query))
+
+    async with app.metadata.transaction(write=True) as tx:
+        tx.execute(f'UPDATE share_grants SET {column}=? WHERE id=?', (value, legacy), write=True)
+    for query in queries:
+        denied = await invoke(app, reader, 'discovery.get', query)
+        assert denied.status == 'error' and denied.error.code == 'permission_denied', wire(denied)
+        assert not denied.resources and not denied.data
+
+    independent = await grant(app, owner, rid, reader[1])
+    for query in queries:
+        ok(await invoke(app, reader, 'discovery.get', query))
+    ok(await invoke(app, owner, 'sharing.revoke', {'grant_id': independent}, version=2))
+    for query in queries:
+        denied = await invoke(app, reader, 'discovery.get', query)
+        assert denied.status == 'error' and denied.error.code == 'permission_denied', wire(denied)
+        assert not denied.resources and not denied.data
diff --git a/tests/test_network_url_ports.py b/tests/test_network_url_ports.py
new file mode 100644
index 00000000..b6b57b55
--- /dev/null
+++ b/tests/test_network_url_ports.py
@@ -0,0 +1,45 @@
+from dataclasses import replace
+
+import pytest
+
+from msg.core.errors import Failure
+from msg.core.models import NetworkPolicy
+from msg.security.network import validate_url
+
+
+@pytest.fixture
+def policy():
+    return NetworkPolicy(
+        schemes=frozenset({'http', 'https'}),
+        hosts=(),
+        ports=frozenset({80, 443}),
+        methods=frozenset({'GET'}),
+        allow_private=False,
+        timeout_ms=1000,
+        max_response_bytes=10000,
+        max_redirects=3,
+    )
+
+
+@pytest.mark.parametrize('scheme', ['http', 'https'])
+def test_explicit_zero_port_does_not_inherit_scheme_default(scheme, policy):
+    with pytest.raises(Failure, match='network_port_forbidden'):
+        validate_url(f'{scheme}://example.org:0/path', 'GET', policy)
+
+
+@pytest.mark.parametrize('port', ['-1', '65536', 'invalid'])
+def test_invalid_explicit_port_is_a_network_url_failure(port, policy):
+    with pytest.raises(Failure, match='invalid_network_url'):
+        validate_url(f'https://example.org:{port}/path', 'GET', policy)
+
+
+@pytest.mark.parametrize(('scheme', 'port'), [('http', 80), ('https', 443)])
+def test_default_port_and_explicit_port_obey_same_allowlist(scheme, port, policy):
+    implicit = f'{scheme}://example.org/path'
+    explicit = f'{scheme}://example.org:{port}/path'
+    assert validate_url(implicit, 'GET', policy).port is None
+    assert validate_url(explicit, 'GET', policy).port == port
+    denied = replace(policy, ports=frozenset({8443}))
+    for url in (implicit, explicit):
+        with pytest.raises(Failure, match='network_port_forbidden'):
+            validate_url(url, 'GET', denied)
diff --git a/tests/test_registry_contract_references.py b/tests/test_registry_contract_references.py
index 55819849..b1ca40f9 100644
--- a/tests/test_registry_contract_references.py
+++ b/tests/test_registry_contract_references.py
@@ -1,6 +1,7 @@
 """Registry freeze rejects dangling references before serving any contract."""
 
-from dataclasses import replace
+from dataclasses import fields, replace
+from types import SimpleNamespace
 
 import pytest
 
@@ -54,6 +55,14 @@ def declarations():
     return resource, capability, operation
 
 
+def contract_adapter(spec, version):
+    # Normal records already reject malformed versions in codec.record.
+    # Registry also accepts plugin-provided objects, so test that boundary
+    # without asking the record constructor to accept an invalid contract.
+    values = {field.name: getattr(spec, field.name) for field in fields(spec)}
+    return SimpleNamespace(**(values | {'version': version}))
+
+
 def registry(resource, capability, operation):
     result = Registry()
     result.add_schema(REF, {'type': 'object'})
@@ -108,3 +117,48 @@ def test_freeze_accepts_registered_content_constraints_and_operation_references(
     assert installed.resource_type('document') == resource
     assert installed.capability('document.read') == capability
     assert installed.operation('document.read') == operation
+
+
+@pytest.mark.parametrize('kind', ['resource_type', 'capability', 'operation'])
+@pytest.mark.parametrize('version', [True, False, 1.0, 1.5, '1', None, 0, -1])
+def test_registration_rejects_non_positive_integer_versions_without_inserting(kind, version):
+    resource, capability, operation = declarations()
+    spec = {'resource_type': resource, 'capability': capability, 'operation': operation}[kind]
+    installed = Registry()
+    register = getattr(installed, 'add_' + kind)
+    with pytest.raises(Failure, match='^invalid_registry_name$'):
+        register(contract_adapter(spec, version))
+    register(spec)
+    assert getattr(installed, kind)(spec.name) == spec
+
+
+@pytest.mark.parametrize('kind', ['resource_type', 'capability', 'operation'])
+@pytest.mark.parametrize('version', [True, 1.0, '1', None])
+def test_manifest_rejects_invalid_versions_before_registering_any_contract(kind, version):
+    resource, capability, operation = declarations()
+    values = {'resource_type': resource, 'capability': capability, 'operation': operation}
+    values[kind] = contract_adapter(values[kind], version)
+    installed = Registry()
+    manifest = PluginManifest(
+        name='identity',
+        version='1',
+        dependencies=(),
+        resource_types=(values['resource_type'],),
+        capabilities=(values['capability'],),
+        operations=(values['operation'],),
+        migrations=(),
+    )
+    with pytest.raises(Failure, match='^invalid_registry_name$'):
+        installed.add(manifest)
+    assert installed.resource_types() == ()
+    assert installed.capabilities() == ()
+    assert installed.operations() == ()
+    installed.add(
+        replace(
+            manifest,
+            resource_types=(resource,),
+            capabilities=(capability,),
+            operations=(operation,),
+        )
+    )
+    assert installed.operation(operation.name) == operation
diff --git a/tests/test_rss_query_validation.py b/tests/test_rss_query_validation.py
new file mode 100644
index 00000000..8dbb4af7
--- /dev/null
+++ b/tests/test_rss_query_validation.py
@@ -0,0 +1,43 @@
+"""RSS reader errors remain client errors on the real HTTP adapter."""
+
+import httpx
+import pytest
+
+from msg.transports.http import create_app
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize('path', ['/rss', '/-/rss'])
+async def test_rss_rejects_nondecimal_digits_without_internal_error(installed, path):
+    app, _ = installed
+    async with app.metadata.transaction(write=False) as tx:
+        before = tx.one('SELECT COUNT(*) FROM events')[0]
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+    ) as http:
+        for limit in ('²', '³', '①', '1²'):
+            response = await http.get(path, params={'limit': limit})
+            assert response.status_code == 400
+            assert response.json()['error']['code'] == 'invalid_limit'
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM events')[0] == before
+
+
+@pytest.mark.asyncio
+async def test_rss_keeps_decimal_limits_and_head_read_only(installed):
+    app, _ = installed
+    async with app.metadata.transaction(write=False) as tx:
+        before = tx.one('SELECT COUNT(*) FROM events')[0]
+    async with httpx.AsyncClient(
+        transport=httpx.ASGITransport(app=create_app(app)), base_url=app.settings.service_url
+    ) as http:
+        for limit in ('1', '01', '١'):
+            response = await http.get('/rss', params={'limit': limit})
+            assert response.status_code == 200
+            assert response.headers['content-type'].startswith('application/rss+xml')
+            head = await http.head('/rss', params={'limit': limit})
+            assert head.status_code == 200
+            assert not head.content
+            assert head.headers['content-length'] == response.headers['content-length']
+    async with app.metadata.transaction(write=False) as tx:
+        assert tx.one('SELECT COUNT(*) FROM events')[0] == before
diff --git a/tests/test_search_phrase_boundaries.py b/tests/test_search_phrase_boundaries.py
new file mode 100644
index 00000000..198978bf
--- /dev/null
+++ b/tests/test_search_phrase_boundaries.py
@@ -0,0 +1,54 @@
+"""Exact phrases must occur within a real searchable field."""
+
+import pytest
+from test_service import call, register
+
+from msg.core.codec import wire
+
+
+@pytest.mark.asyncio
+async def test_exact_phrase_does_not_cross_name_body_boundary(installed):
+    app, _ = installed
+    key, subject, _ = await register(app, 'phrase-boundary-owner')
+    split = await call(
+        app,
+        'content.post_create',
+        {'parent': '/main', 'name': 'boundaryalpha', 'body': 'boundarybeta'},
+        key=key,
+        subject=subject,
+    )
+    complete = await call(
+        app,
+        'content.post_create',
+        {'parent': '/main', 'name': 'complete', 'body': 'boundaryalpha.md boundarybeta'},
+        key=key,
+        subject=subject,
+    )
+    assert split.status == complete.status == 'ok'
+    result = await call(
+        app,
+        'discovery.lexical_search',
+        {
+            'scope': '/main',
+            'exact': 'boundaryalpha.md boundarybeta',
+            'snippet': True,
+            'explain': 'compact',
+        },
+    )
+    assert result.status == 'ok', wire(result)
+    assert [item['id'] for item in result.data['items']] == [complete.resources[0].id]
+    item = result.data['items'][0]
+    assert item['snippet']['field'] == 'body'
+    assert list(item['rank_reason']['matched_fields']) == ['body']
+
+    # Independent terms can still match different fields of the same resource.
+    terms = await call(
+        app,
+        'discovery.lexical_search',
+        {'scope': '/main', 'terms': 'boundaryalpha boundarybeta', 'mode': 'all'},
+    )
+    assert terms.status == 'ok', wire(terms)
+    assert {item['id'] for item in terms.data['items']} == {
+        split.resources[0].id,
+        complete.resources[0].id,
+    }
diff --git a/tests/test_text_patch_eof_boundaries.py b/tests/test_text_patch_eof_boundaries.py
new file mode 100644
index 00000000..6ee73548
--- /dev/null
+++ b/tests/test_text_patch_eof_boundaries.py
@@ -0,0 +1,117 @@
+"""Unified patches cannot turn an EOF marker into an interior byte join."""
+
+import subprocess
+
+import pytest
+
+from msg.core.errors import Failure
+from msg.core.text_patch import apply_patch
+
+
+@pytest.mark.parametrize('rebase', [False, True])
+def test_new_eof_marker_rejects_remaining_source_lines_like_git(tmp_path, rebase):
+    source = 'a\nuntouched\n'
+    diff = '--- a/f\n+++ b/f\n@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n'
+    target = tmp_path / 'f'
+    target.write_bytes(source.encode())
+    result = subprocess.run(
+        ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), capture_output=True
+    )
+    assert result.returncode != 0
+    assert target.read_bytes() == source.encode()
+    kwargs = {'base_source': source} if rebase else {}
+    with pytest.raises(Failure) as caught:
+        apply_patch(source, {'kind': 'unified', 'diff': diff}, **kwargs)
+    assert caught.value.code == 'invalid_unified_patch'
+
+
+def test_unterminated_diff_payload_requires_explicit_eof_marker_like_git(tmp_path):
+    source = 'a\n'
+    diff = '--- a/f\n+++ b/f\n@@ -1 +1 @@\n-a\n+b'
+    target = tmp_path / 'f'
+    target.write_bytes(source.encode())
+    result = subprocess.run(
+        ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), capture_output=True
+    )
+    assert result.returncode != 0
+    assert target.read_bytes() == source.encode()
+    with pytest.raises(Failure) as caught:
+        apply_patch(source, {'kind': 'unified', 'diff': diff})
+    assert caught.value.code == 'invalid_unified_patch'
+
+
+@pytest.mark.parametrize('rebase', [False, True])
+def test_real_eof_marker_still_matches_git(tmp_path, rebase):
+    source = 'prefix\na\n'
+    diff = '--- a/f\n+++ b/f\n@@ -2 +2 @@\n-a\n+b\n\\ No newline at end of file\n'
+    target = tmp_path / 'f'
+    target.write_bytes(source.encode())
+    subprocess.run(
+        ['git', 'apply', '-'], cwd=tmp_path, input=diff.encode(), check=True, capture_output=True
+    )
+    kwargs = {'base_source': source} if rebase else {}
+    assert (
+        apply_patch(source, {'kind': 'unified', 'diff': diff}, **kwargs).encode()
+        == target.read_bytes()
+    )
+    assert target.read_bytes() == b'prefix\nb'
+
+
+def test_rebase_does_not_join_concurrently_appended_lines():
+    base = 'a\n'
+    current = base + 'concurrent\n'
+    diff = '@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n'
+    assert apply_patch(base, {'kind': 'unified', 'diff': diff}) == 'b'
+    with pytest.raises(Failure) as caught:
+        apply_patch(current, {'kind': 'unified', 'diff': diff}, base_source=base)
+    assert caught.value.code == 'invalid_unified_patch'
+
+
+@pytest.mark.asyncio
+async def test_signed_patch_failure_does_not_publish_revision(installed):
+    from test_service import call, register
+
+    from msg.core.codec import wire
+
+    app, _ = installed
+    key, owner, _ = await register(app, 'patch-eof-owner')
+    post = await call(
+        app,
+        'content.post_create',
+        {'parent': '/main', 'body': 'a\nuntouched\n'},
+        key=key,
+        subject=owner,
+    )
+    assert post.status == 'ok', wire(post)
+    rid = post.resources[0].id
+
+    async def snapshot():
+        async with app.metadata.transaction(write=False) as tx:
+            resource = await tx.resource(rid)
+            return (
+                resource.generation,
+                resource.revision,
+                tx.one('SELECT COUNT(*) FROM revisions')[0],
+                tx.one('SELECT COUNT(*) FROM events')[0],
+            )
+
+    before = await snapshot()
+    failed = await call(
+        app,
+        'content.text_patch',
+        {
+            'id': rid,
+            'base_revision': post.resources[0].revision,
+            'base_generation': post.data['generation'],
+            'patch': {
+                'kind': 'unified',
+                'diff': '@@ -1 +1 @@\n-a\n+b\n\\ No newline at end of file\n',
+            },
+        },
+        key=key,
+        subject=owner,
+        expected=((rid, post.data['generation']),),
+        contract_version=3,
+    )
+    assert failed.status == 'error' and failed.error.code == 'invalid_unified_patch', wire(failed)
+    assert await snapshot() == before
diff --git a/tests/test_webhook_delivery.py b/tests/test_webhook_delivery.py
index 37f7818a..6f1d9765 100644
--- a/tests/test_webhook_delivery.py
+++ b/tests/test_webhook_delivery.py
@@ -11,7 +11,13 @@
 from msg.core.codec import b64, canonical, wire
 from msg.core.errors import Failure
 from msg.workers.effects import EffectWorker
-from msg.workers.webhook import PublicResolver, sign_delivery, validate_endpoint, verify_delivery
+from msg.workers.webhook import (
+    PublicResolver,
+    WebhookSender,
+    sign_delivery,
+    validate_endpoint,
+    verify_delivery,
+)
 
 
 class Sender:
@@ -196,6 +202,8 @@ async def send(self, *args, **kwargs):
         'https://[::1]/hook',
         'https://10.1.2.3/hook',
         'https://example.org:8443/hook',
+        'https://example.org:0/hook',
+        'https://example.org:000/hook',
         'https://user:pass@example.org/hook',
     ],
 )
@@ -204,6 +212,28 @@ def test_webhook_rejects_unsafe_endpoint(url):
         validate_endpoint(url)
 
 
+@pytest.mark.parametrize('authority', ['example.org', 'example.org:443'])
+def test_webhook_accepts_only_default_or_explicit_https_port(authority):
+    assert validate_endpoint(f'https://{authority}/hook') == ('example.org', 443)
+
+
+@pytest.mark.asyncio
+async def test_webhook_zero_port_rejected_before_network_setup(monkeypatch):
+    def unexpected_connector(*args, **kwargs):
+        pytest.fail('invalid endpoint reached network setup')
+
+    monkeypatch.setattr('msg.workers.webhook.aiohttp.TCPConnector', unexpected_connector)
+    with pytest.raises(Failure, match='webhook_port_forbidden'):
+        await WebhookSender().send(
+            'https://example.org:0/hook',
+            b'x' * 32,
+            b'{}',
+            timestamp='1',
+            event_id='e_a',
+            delivery_id='job_a',
+        )
+
+
 def test_webhook_signature_covers_raw_body_and_timestamp():
     secret = secrets.token_bytes(32)
     timestamp = str(int(NOW.timestamp()))