Add CLI read-field discovery and bounded local output formats

docs/CLI_FIELD_SELECTION.md

@@ -0,0 +1,85 @@
+# CLI read field selection
+
+`msg call` supports an explicit `--contract-version` (default `1`, preserving
+existing calls). `--json` is opt-in and uses the selected server Registry
+contract. It is not a client-side truncation of a full business response.
+
+Discover ReadQuery v3 fields without executing the business query:
+
+```sh
+msg call discovery.read_query --contract-version 3 --json
+```
+
+Select a bounded page using signed server-side projection:
+
+```sh
+msg call discovery.read_query '{"parent":"/main","query_version":3,"limit":20}' \
+  --contract-version 3 --json id,name,path
+```
+
+The unchanged JSON result envelope retains the cursor/next links and status.
+For the next page, send only the returned cursor at the same contract version,
+without repeating `--json`; the server-bound cursor retains the projection:
+
+```sh
+msg call discovery.read_query '{"cursor":"SERVER_RETURNED_CURSOR"}' --contract-version 3
+```
+
+A bare `--json` lists the `fields.items.enum` from the exact registered input
+schema and does not enumerate resources. Contracts without this declared field
+vocabulary return `json_fields_unavailable`; the client does not guess fields
+from one object's data or import the server runtime. Non-read contracts return
+`json_read_required` before business execution. Existing `fields`, a cursor, or
+`--return-field` with `--json` returns `json_query_conflict`. Unknown, repeated,
+empty, or whitespace-altered field names return `invalid_json_fields`.
+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.
+Only the returned authorized JSON envelope is processed, after a successful read.
+Server error envelopes remain unchanged and return a nonzero status.
+
+```sh
+msg call discovery.read_query '{"parent":"/main","query_version":3,"limit":20}' \
+  --contract-version 3 --json id,name --jq '.data.items[] | .name'
+msg call discovery.read_query '{"parent":"/main","query_version":3,"limit":1}' \
+  --contract-version 3 --json id,name --template 'Name: {{/data/items/0/name}}'
+```
+
+`--jq` uses an installed `jq` executable directly with an argument vector, never
+a shell. It produces compact JSON values, one per line. No executable gives
+`jq_unavailable`; invalid filters give `invalid_jq_query`. The process receives
+an empty environment and an empty temporary working/module directory. Filters
+containing the words `import`, `include`, or `modulemeta` are conservatively
+rejected, even inside quoted strings or comments: module loading is unsupported.
+No other command-line options or filenames are passed to jq. Query text is
+limited to 4 KiB; input and output to 1 MiB; execution to 2 seconds. Output is
+bounded while streaming, and overflow/timeout kills and waits for the child.
+Errors are stable codes and never include jq diagnostics or private input.
+These are I/O and elapsed-time limits, not a general resource sandbox: jq
+intermediate allocations remain subject to the host's process memory limits.
+Run only filters you intentionally choose; server content is always data.
+
+`--template` supports only `{{JSON_POINTER}}` placeholders. JSON Pointer follows
+object keys and array indices, never Python attributes: `/data/items/0/name`
+selects that JSON value; `~1` escapes `/`, and `~0` escapes `~` in an object key.
+`{{}}` selects the entire result. Values are substituted as JSON, so a string
+is quoted and its terminal control characters are escaped. Literal template
+text is preserved; the CLI appends a newline. There are no loops, expressions,
+attribute traversal, template files or external access. Malformed placeholders,
+invalid escapes, and missing pointers give `invalid_output_template`, not empty
+success. Template text is limited to 4 KiB and the result to 1 MiB.
+
+This slice does not implement global output flags on every shortcut or automatic
+pagination. Existing `read --field`, `search --fields`, mutation calls,
+authentication and signing stay intact. Default output remains the unchanged
+JSON envelope; templates explicitly opt into literal text output. Issue #82
+remains open for the other command and pagination contracts.
+
+## Test prerequisites
+
+The real formatter regression tests require `jq` on `PATH`; absence fails rather
+than skipping those assertions. The signed HTTP/read-only test additionally
+requires the standard disposable PostgreSQL fixture from `CONTRIBUTING.md`.

src/msg/cli.py

@@ -106,6 +106,20 @@ def parser():
     )
     call.add_argument('operation')
     call.add_argument('arguments', nargs='?', default='{}')
+    call.add_argument('--contract-version', type=int, default=1)
+    call.add_argument(
+        '--json',
+        dest='json_fields',
+        nargs='?',
+        const='',
+        metavar='FIELDS',
+        help='Select server-declared read fields; without FIELDS, list them without reading data.',
+    )
+    output_format = call.add_mutually_exclusive_group()
+    output_format.add_argument(
+        '--jq', help='Run a bounded local jq filter on selected read output.'
+    )
+    output_format.add_argument('--template', help='Local data-only JSON-pointer template.')
     call.add_argument('--request-id')
     call.add_argument('--expect', action='append', default=[], metavar='ID=GENERATION')
     call.add_argument('--return-field', action='append', default=[])
@@ -463,13 +477,30 @@ async def run(args):
                 rid, sep, generation = item.rpartition('=')
                 require(sep and generation.isdecimal(), 'invalid_expected_generation')
                 expected.append((rid, int(generation)))
-            result = await client.call(
-                args.operation,
-                arguments(args.arguments),
-                request_id=args.request_id,
-                expected=expected,
-                return_fields=args.return_field,
+            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
+            if args.json_fields is not None:
+                from msg.client_output import json_fields
+
+                require(not args.return_field, 'json_query_conflict')
+                params, result = await json_fields(
+                    client, args.operation, args.contract_version, params, args.json_fields
+                )
+            if result is None:
+                result = await client.call(
+                    args.operation,
+                    params,
+                    request_id=args.request_id,
+                    expected=expected,
+                    return_fields=args.return_field,
+                    contract_version=args.contract_version,
+                )
         elif command in {'money', 'store', 'bounty', 'orders', 'delivery'}:
             from msg.client_market import run_command
 
@@ -923,7 +954,19 @@ async def run(args):
             and rendered.get('status') == 'ok'
         ):
             rendered['data'].pop('url', None)
-        print(canonical(rendered).decode())
+        if command == 'call' and not (
+            isinstance(result, OperationResult) and result.status != 'ok'
+        ):
+            from msg.client_output import render_jq, render_template
+
+            if args.jq is not None:
+                print(await render_jq(args.jq, rendered), end='')
+            elif args.template is not None:
+                print(render_template(args.template, rendered))
+            else:
+                print(canonical(rendered).decode())
+        else:
+            print(canonical(rendered).decode())
         return 1 if isinstance(result, OperationResult) and result.status == 'error' else 0
     finally:
         await transport.close()

src/msg/client_output.py

@@ -0,0 +1,167 @@
+"""Read field selection comes from the exact server Registry contract."""
+
+import asyncio
+import re
+import shutil
+import tempfile
+from collections.abc import Mapping
+from contextlib import suppress
+
+from msg.core.codec import canonical
+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')
+    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
+    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')
+    schema = data.get('input')
+    require(isinstance(schema, Mapping), 'json_fields_unavailable')
+    properties = schema.get('properties')
+    require(isinstance(properties, Mapping), 'json_fields_unavailable')
+    field = properties.get('fields')
+    require(isinstance(field, Mapping), 'json_fields_unavailable')
+    items = field.get('items')
+    require(isinstance(items, Mapping), 'json_fields_unavailable')
+    available = items.get('enum')
+    require(
+        isinstance(available, (list, tuple))
+        and available
+        and all(isinstance(name, str) and name for name in available),
+        'json_fields_unavailable',
+    )
+    if selection == '':
+        return None, {'status': 'ok', 'data': {'fields': list(available)}}
+    chosen = selection.split(',')
+    require(
+        len(chosen) == len(set(chosen)) and set(chosen) <= set(available),
+        'invalid_json_fields',
+    )
+    return {**arguments, 'fields': chosen}, None
+
+
+MAX_FORMAT_BYTES = 1048576
+MAX_EXPRESSION_BYTES = 4096
+_POINTER = re.compile(r'\{\{(/[^{}]*)?\}\}')
+
+
+def _bounded(value, maximum=MAX_FORMAT_BYTES):
+    require(len(value.encode('utf-8')) <= maximum, 'output_format_too_large')
+    return value
+
+
+def render_template(template, data):
+    """Replace JSON-pointer placeholders with JSON values; never evaluate text."""
+    _bounded(template, MAX_EXPRESSION_BYTES)
+    parts = []
+    offset = 0
+    size = 0
+    for match in _POINTER.finditer(template):
+        literal = template[offset : match.start()]
+        require('{{' not in literal and '}}' not in literal, 'invalid_output_template')
+        value = data
+        pointer = match.group(1)
+        for token in pointer.split('/')[1:] if pointer is not None else ():
+            require(not re.search(r'~(?![01])', token), 'invalid_output_template')
+            key = token.replace('~1', '/').replace('~0', '~')
+            if isinstance(value, Mapping):
+                require(key in value, 'invalid_output_template')
+                value = value[key]
+            elif isinstance(value, (list, tuple)):
+                require(
+                    re.fullmatch(r'0|[1-9][0-9]{0,8}', key) and int(key) < len(value),
+                    'invalid_output_template',
+                )
+                value = value[int(key)]
+            else:
+                raise Failure('invalid_output_template')
+        replacement = canonical(value).decode()
+        size += len((literal + replacement).encode())
+        require(size <= MAX_FORMAT_BYTES, 'output_format_too_large')
+        parts.extend((literal, replacement))
+        offset = match.end()
+    tail = template[offset:]
+    require('{{' not in tail and '}}' not in tail, 'invalid_output_template')
+    parts.append(tail)
+    return _bounded(''.join(parts))
+
+
+async def render_jq(query, data, *, timeout=2):
+    """Run optional jq on one returned JSON value, with no inherited secrets."""
+    _bounded(query, MAX_EXPRESSION_BYTES)
+    # Deliberately conservative, including quoted occurrences: jq modules may
+    # load filesystem data. This interface supports filters, not module loading.
+    require(not re.search(r'\b(?:import|include|modulemeta)\b', query), 'invalid_jq_query')
+    executable = shutil.which('jq')
+    require(executable is not None, 'jq_unavailable')
+    payload = canonical(data) + b'\n'
+    require(len(payload) <= MAX_FORMAT_BYTES, 'output_format_too_large')
+    with tempfile.TemporaryDirectory(prefix='msg-jq-') as directory:
+        try:
+            process = await asyncio.create_subprocess_exec(
+                executable,
+                '-c',
+                '-M',
+                '-L',
+                directory,
+                '--',
+                query,
+                stdin=asyncio.subprocess.PIPE,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.DEVNULL,
+                cwd=directory,
+                env={},
+            )
+        except OSError:
+            raise Failure('jq_unavailable') from None
+
+        async def feed():
+            try:
+                process.stdin.write(payload)
+                await process.stdin.drain()
+            except BrokenPipeError, ConnectionResetError:
+                pass
+            finally:
+                process.stdin.close()
+                with suppress(BrokenPipeError, ConnectionResetError):
+                    await process.stdin.wait_closed()
+
+        async def collect():
+            chunks = []
+            size = 0
+            while block := await process.stdout.read(65536):
+                size += len(block)
+                require(size <= MAX_FORMAT_BYTES, 'output_format_too_large')
+                chunks.append(block)
+            return b''.join(chunks)
+
+        tasks = [
+            asyncio.create_task(feed()),
+            asyncio.create_task(collect()),
+            asyncio.create_task(process.wait()),
+        ]
+        try:
+            _, output, status = await asyncio.wait_for(asyncio.gather(*tasks), timeout)
+            require(status == 0, 'invalid_jq_query')
+            try:
+                return output.decode('utf-8')
+            except UnicodeError:
+                raise Failure('invalid_jq_output') from None
+        except TimeoutError:
+            raise Failure('output_format_timeout') from None
+        finally:
+            if process.returncode is None:
+                with suppress(ProcessLookupError):
+                    process.kill()
+            for task in tasks:
+                if not task.done():
+                    task.cancel()
+            await asyncio.gather(*tasks, return_exceptions=True)
+            await process.wait()