Preserve CLI formatting on ReadQuery cursor continuation

docs/CLI_FIELD_SELECTION.md

@@ -36,8 +36,24 @@ Server schema failures retain the normal error result and nonzero exit status.
 
 ## Optional local output processing
 
-Both options require explicit nonempty `--json FIELDS` on `msg call`; they are
-mutually exclusive. Bare field discovery cannot be combined with formatting.
+Both options require explicit nonempty `--json FIELDS` on the first `msg call`
+read; they are mutually exclusive. Bare field discovery cannot be combined
+with formatting. A canonical `discovery.read_query` continuation can instead
+format its cursor-bound fields without a new `--json` projection:
+
+```sh
+msg call discovery.read_query '{"cursor":"SERVER_RETURNED_CURSOR"}' \
+  --contract-version 3 --jq '.data.items[] | .name'
+```
+
+The continuation must contain only `cursor`, preserve the original contract
+version, and omit `--json` and `--return-field`. The client verifies the exact
+read-only operation/version from Registry again, then signs the unchanged
+cursor arguments. The server rechecks the cursor and current access; the client
+does not infer or change its projection or automatically follow another page.
+Other operations and mixed cursor/query arguments are rejected before network
+access. Schema errors remain normal error results without formatting.
+
 Only the returned authorized JSON envelope is processed, after a successful read.
 Server error envelopes remain unchanged and return a nonzero status.
 

src/msg/cli.py

@@ -477,14 +477,17 @@ async def run(args):
                 rid, sep, generation = item.rpartition('=')
                 require(sep and generation.isdecimal(), 'invalid_expected_generation')
                 expected.append((rid, int(generation)))
-            require(
-                not (args.jq is not None or args.template is not None)
-                or args.json_fields not in {None, ''},
-                'json_fields_required',
-            )
             params = arguments(args.arguments)
             require(args.contract_version > 0, 'invalid_operation_version')
             result = None
+            formatted = args.jq is not None or args.template is not None
+            if formatted and args.json_fields is None and 'cursor' in params:
+                from msg.client_output import cursor_output
+
+                require(not args.return_field, 'json_query_conflict')
+                result = await cursor_output(client, args.operation, args.contract_version, params)
+            else:
+                require(not formatted or args.json_fields not in {None, ''}, 'json_fields_required')
             if args.json_fields is not None:
                 from msg.client_output import json_fields
 

src/msg/client_output.py

@@ -11,17 +11,43 @@
 from msg.core.errors import Failure, require
 
 
-async def json_fields(client, operation, version, arguments, selection):
-    """Return signed-query arguments or a discovery/error result, never both."""
-    require('fields' not in arguments and 'cursor' not in arguments, 'json_query_conflict')
+async def read_contract(client, operation, version):
+    """Discover the exact version and preserve unsuccessful server responses."""
     require(type(version) is int and version > 0, 'invalid_operation_version')
     response = await client.call('discovery.schema', {'operation': f'{operation}@{version}'})
     if response.status != 'ok':
-        return None, response
+        return response
     data = response.data
     require(isinstance(data, Mapping), 'invalid_read_response')
     spec = data.get('operation')
     require(isinstance(spec, Mapping) and spec.get('effect') == 'read', 'json_read_required')
+    require(
+        spec.get('name') == operation and spec.get('version') == version,
+        'invalid_read_response',
+    )
+    return response
+
+
+async def cursor_output(client, operation, version, arguments):
+    """Validate formatting for a bound ReadQuery cursor without a new projection."""
+    require(operation == 'discovery.read_query', 'cursor_output_read_required')
+    require(
+        set(arguments) == {'cursor'}
+        and isinstance(arguments['cursor'], str)
+        and bool(arguments['cursor']),
+        'cursor_query_mismatch',
+    )
+    response = await read_contract(client, operation, version)
+    return None if response.status == 'ok' else response
+
+
+async def json_fields(client, operation, version, arguments, selection):
+    """Return signed-query arguments or a discovery/error result, never both."""
+    require('fields' not in arguments and 'cursor' not in arguments, 'json_query_conflict')
+    response = await read_contract(client, operation, version)
+    if response.status != 'ok':
+        return None, response
+    data = response.data
     schema = data.get('input')
     require(isinstance(schema, Mapping), 'json_fields_unavailable')
     properties = schema.get('properties')

tests/test_cli_cursor_output.py

@@ -0,0 +1,147 @@
+"""Local output formatting must survive a server-bound cursor continuation."""
+
+import pytest
+
+from msg import cli
+from msg.core.codec import loads
+from msg.core.errors import Failure
+from msg.core.models import OperationResult
+
+
+@pytest.fixture
+def cursor_cli(tmp_path, monkeypatch):
+    calls = []
+
+    class Transport:
+        async def close(self):
+            pass
+
+    class Client:
+        def __init__(self, state, transport):
+            pass
+
+        async def call(self, operation, arguments, **options):
+            calls.append((operation, arguments, options))
+            data = (
+                {'operation': {'effect': 'read', 'name': 'discovery.read_query', 'version': 3}}
+                if operation == 'discovery.schema'
+                else {'items': [{'name': 'next-page'}], 'cursor': 'next'}
+            )
+            return OperationResult(
+                request_id='read',
+                operation=operation,
+                status='ok',
+                actor=None,
+                subject=None,
+                data=data,
+            )
+
+    monkeypatch.setitem(cli.TRANSPORTS, 'http', lambda server: Transport())
+    monkeypatch.setattr(cli, 'MsgClient', Client)
+
+    async def invoke(operation, arguments, *options):
+        return await cli.run(
+            cli.parser().parse_args([
+                '--config-dir',
+                str(tmp_path),
+                'call',
+                operation,
+                arguments,
+                '--contract-version',
+                '3',
+                *options,
+            ])
+        )
+
+    return invoke, calls
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('flag', 'expression'),
+    [
+        ('--jq', '.data.items[0].name'),
+        ('--template', '{{/data/items/0/name}}'),
+    ],
+)
+async def test_cursor_format_keeps_exact_signed_cursor_and_version(
+    cursor_cli, capsys, flag, expression
+):
+    invoke, calls = cursor_cli
+    assert await invoke('discovery.read_query', '{"cursor":"opaque"}', flag, expression) == 0
+    captured = capsys.readouterr()
+    assert captured.err == '' and loads(captured.out.encode()) == 'next-page'
+    assert calls[0][:2] == ('discovery.schema', {'operation': 'discovery.read_query@3'})
+    assert calls[1][0:2] == ('discovery.read_query', {'cursor': 'opaque'})
+    assert calls[1][2]['contract_version'] == 3
+    assert len(calls) == 2
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ('operation', 'arguments', 'options'),
+    [
+        ('content.post_create', '{"cursor":"opaque"}', []),
+        ('discovery.get', '{"cursor":"opaque"}', []),
+        ('discovery.read_query', '{"cursor":"opaque","fields":["id"]}', []),
+        ('discovery.read_query', '{"cursor":"opaque","limit":10}', []),
+        ('discovery.read_query', '{"cursor":""}', []),
+        ('discovery.read_query', '{"cursor":null}', []),
+        ('discovery.read_query', '{"cursor":0}', []),
+        ('discovery.read_query', '{"cursor":"opaque"}', ['--return-field', 'status']),
+    ],
+)
+async def test_cursor_format_rejects_other_operations_or_conflicts_before_network(
+    cursor_cli, operation, arguments, options
+):
+    invoke, calls = cursor_cli
+    with pytest.raises(
+        Failure, match='cursor_output_read_required|cursor_query_mismatch|json_query_conflict'
+    ):
+        await invoke(operation, arguments, '--template', '{{/data}}', *options)
+    assert calls == []
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    'spec',
+    [
+        {'name': 'discovery.get', 'version': 3, 'effect': 'read'},
+        {'name': 'discovery.read_query', 'version': 1, 'effect': 'read'},
+        {'name': 'discovery.read_query', 'version': 3, 'effect': 'transaction'},
+    ],
+)
+async def test_cursor_output_refuses_changed_contract_identity_or_effect(spec):
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    from msg.client_output import cursor_output
+
+    client = SimpleNamespace(
+        call=AsyncMock(
+            return_value=SimpleNamespace(
+                status='ok',
+                data={'operation': spec},
+            )
+        )
+    )
+    with pytest.raises(Failure, match='invalid_read_response|json_read_required'):
+        await cursor_output(client, 'discovery.read_query', 3, {'cursor': 'opaque'})
+    client.call.assert_awaited_once_with(
+        'discovery.schema', {'operation': 'discovery.read_query@3'}
+    )
+
+
+@pytest.mark.asyncio
+async def test_cursor_output_preserves_unsuccessful_schema_result():
+    from types import SimpleNamespace
+    from unittest.mock import AsyncMock
+
+    from msg.client_output import cursor_output
+
+    error = SimpleNamespace(status='error')
+    client = SimpleNamespace(call=AsyncMock(return_value=error))
+    assert await cursor_output(client, 'discovery.read_query', 3, {'cursor': 'opaque'}) is error
+    client.call.assert_awaited_once_with(
+        'discovery.schema', {'operation': 'discovery.read_query@3'}
+    )

tests/test_cli_json_fields.py

@@ -28,13 +28,15 @@ def __init__(self, *, effect='read', fields=('id', 'name'), status='ok'):
         self.result = SimpleNamespace(
             status=status,
             data={
-                'operation': {'effect': effect},
+                'operation': {'effect': effect, 'name': 'discovery.read_query', 'version': 3},
                 'input': {'properties': {'fields': {'items': {'enum': list(fields)}}}},
             },
         )
 
     async def call(self, operation, arguments):
         self.calls.append((operation, arguments))
+        name, _, version = arguments['operation'].rpartition('@')
+        self.result.data['operation'].update(name=name, version=int(version))
         return self.result
 
 
@@ -196,6 +198,19 @@ async def invoke(operation, arguments, *flags):
             )
             assert status == 0 and set(next_page['data']['items'][0]) == {'id', 'name'}
             assert next_page['data']['items'][0]['id'] != page['data']['items'][0]['id']
+            for flag, expression in (
+                ('--jq', '.data.items[0].name'),
+                ('--template', '{{/data/items/0/name}}'),
+            ):
+                status, name = await invoke(
+                    'discovery.read_query',
+                    {'cursor': page['data']['cursor']},
+                    '--contract-version',
+                    '3',
+                    flag,
+                    expression,
+                )
+                assert status == 0 and name == next_page['data']['items'][0]['name']
             with pytest.raises(Failure, match='json_query_conflict'):
                 await invoke(
                     'discovery.read_query',