diff --git a/src/agent_comms/channel_targets.py b/src/agent_comms/channel_targets.py
index 64aa313f4b..3e13a6f9e8 100644
--- a/src/agent_comms/channel_targets.py
+++ b/src/agent_comms/channel_targets.py
@@ -22,10 +22,9 @@
 if TYPE_CHECKING:
     from .audience_manifest import AudiencedMessage
     from .frozen_recipient import FrozenRecipient
-    from .catalog_document import CatalogAccess, CatalogDocument
+    from .catalog_document import CatalogDocument
     from .channels import Channel
     from .cli_command import CliCommand
-    from .cli_result import CliResult
     from .comms import Comms
     from .goal_management import Goals
     from .historical_views import ChannelHistory, DMHistory, HistoryView
@@ -163,11 +162,6 @@ def available(self, snapshot: RegistrySnapshot) -> bool:
     @abstractmethod
     def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], Channel]): ...
 
-    @abstractmethod
-    def bindings[C: CliCommand](
-        self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved,
-        *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]: ...
 
     def history_view(self, viewer: DisplayViewer, require_channel: Callable[[str], Channel],
                      channel: type[ChannelHistory], direct: type[DMHistory]) -> HistoryView:
@@ -178,7 +172,7 @@ def history_result(self, require_channel: Callable[[str], Channel], messages: tu
         raise ValueError("History requires a #channel or DM peer.")
 
     def contexts(
-        self, declaration: type[CliCommand[CliResult]], channel: ChannelSelection,
+        self, declaration: type[CliCommand], channel: ChannelSelection,
     ) -> tuple[SurroundingChannel, ...]:
         return (NoChannel(),)
 
@@ -394,7 +388,7 @@ def history_result(self, require_channel: Callable[[str], Channel], messages: tu
         return self.history_result_type.for_target(self.name, messages, has_older)
 
     def contexts(
-        self, declaration: type[CliCommand[CliResult]], channel: ChannelSelection,
+        self, declaration: type[CliCommand], channel: ChannelSelection,
     ) -> tuple[SurroundingChannel, ...]:
         return declaration.selection_channels(channel, self.name)
 
@@ -411,11 +405,6 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
         thread = snapshot.require(self.selected_name(snapshot))
         return ResolvedThread(thread, snapshot.status(thread.name))
 
-    def bindings[C: CliCommand](
-        self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved: ResolvedThread,
-        *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
-        return declaration.thread_bindings(comms, resolved.thread, resolved.status, channel, catalog=catalog)
 
     def validate(self, message: Message) -> None:
         if message.sender == self.name and message.type.conversation_message:
@@ -560,11 +549,6 @@ def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], C
                         if channel.matches(thread.tags))
         return ResolvedChannel(channel, members)
 
-    def bindings[C: CliCommand](
-        self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved: ResolvedChannel,
-        *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
-        return declaration.channel_bindings(comms, resolved, project=project, snapshot=snapshot, catalog=catalog)
 
     def validate(self, sender: Message | None = None):
         from .channels import BuiltinChannelView
@@ -652,11 +636,6 @@ def destination_names(self) -> tuple[str, ...]:
     def resolve(self, snapshot: RegistrySnapshot, require_channel: Callable[[str], Channel]):
         return self.project
 
-    def bindings[C: CliCommand](
-        self, declaration: type[C], comms: Comms, channel: SurroundingChannel, resolved,
-        *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-    ) -> tuple[C, ...]:
-        return declaration.project_bindings(comms, resolved)
 
 
 _TAG_CHARS = set("abcdefghijklmnopqrstuvwxyz0123456789-_")
diff --git a/src/agent_comms/cli_command.py b/src/agent_comms/cli_command.py
index 6af0987dee..95ed04b075 100644
--- a/src/agent_comms/cli_command.py
+++ b/src/agent_comms/cli_command.py
@@ -15,6 +15,7 @@
 from agent_comms.lifecycle_capability import LifecycleCapability, Successful
 from agent_comms.registry_document import RegistrySnapshot
 from agent_comms.target_fields import TargetField
+from agent_comms.thread_model import CatalogSelection
 from dataclasses import dataclass, field, replace
 from agent_comms_hooks import ToolRefusedError
 from pydantic import JsonValue
@@ -26,8 +27,9 @@
 from agent_comms.nominal.declared_family import ModuleMembers
 from agent_comms.nominal.field_codec import FieldPlan, TargetBound
 from agent_comms.launch_root import LaunchRoot
+from functools import singledispatchmethod
 from inspect import iscoroutinefunction, signature
-from types import get_original_bases
+from types import MappingProxyType, get_original_bases
 from typing import TYPE_CHECKING, TextIO, cast, get_args, get_origin
 
 if TYPE_CHECKING:
@@ -63,15 +65,156 @@ def __init_subclass__(cls) -> None:
 Answer = TypeVar("Answer", bound=CliResult, covariant=True, default=CliResult)
 
 
+class CommandBinding(ABC):
+    """A declared operation can collect fields before a complete command exists."""
+
+    @property
+    @abstractmethod
+    def declaration(self) -> type[CliCommand]: ...
+
+    @property
+    def editable_fields(self) -> tuple[TargetField, ...]:
+        return tuple(self.field_value(entry)
+                     for entry in FieldCodec.argument_fields(self.declaration)
+                     if not entry.setting(TargetBound, False))
+
+    @abstractmethod
+    def field_value(self, entry: FieldPlan) -> TargetField: ...
+
+    @abstractmethod
+    def retained_payload(self) -> dict[str, object]:
+        """Already decoded constructor values, never serialized for editing."""
+
+    def for_editor(self, comms: Comms, target: str, project: Checkout) -> Self:
+        return self
+
+    def for_fields(self, arguments: Mapping[str, object]) -> CommandBinding:
+        return self.edited(self.declaration.editor_arguments(arguments)) if arguments else self
+
+    async def prepare_field(self, item: TargetField, catalogs: ConfigOptions) -> TargetField:
+        return await item.prepare(catalogs)
+
+    async def prepare_fields(
+        self, arguments: Mapping[str, object] = MappingProxyType({}),
+    ) -> tuple[TargetField, ...]:
+        binding = self.for_fields(arguments)
+        catalogs = ConfigOptions.configured()
+        return tuple([await binding.prepare_field(item, catalogs)
+                      for item in binding.editable_fields])
+
+    def authorize_edits(self, arguments: Mapping[str, object]) -> frozenset[str]:
+        bound = frozenset(entry.key for entry in FieldCodec.argument_fields(self.declaration)
+                          if entry.setting(TargetBound, False))
+        if bound & arguments.keys():
+            raise ValueError("Target-bound parameters cannot be overridden")
+        if arguments.keys() - {item.name for item in self.editable_fields}:
+            raise ValueError("Parameters not offered by this target command cannot be overridden")
+        return bound
+
+    @staticmethod
+    def decoded_edits(declaration: type, arguments: Mapping[str, object]) -> dict[str, object]:
+        return {name: value
+                for entry in FieldCodec.argument_fields(declaration) if entry.key in arguments
+                for name, value in entry.decoded(arguments).items()}
+
+    def edited(self, arguments: Mapping[str, object]) -> CliCommand:
+        self.authorize_edits(arguments)
+        values = {**self.retained_payload(), **self.decoded_edits(self.declaration, arguments)}
+        for parameter in self.editable_fields:
+            parameter.plan.require_constructor_input(values)
+        return FieldCodec.construct(self.declaration, values)
+
+    @abstractmethod
+    def confirmation(self) -> str: ...
+
+    @classmethod
+    @abstractmethod
+    def selection_confirmation(cls, commands: tuple[Self, ...]) -> str: ...
+
+    @abstractmethod
+    def with_confirmation(self, confirmed: bool) -> CommandBinding: ...
+
+
+class CatalogBinding(CommandBinding):
+    """Only a binding declaring catalog inputs has a catalog question to answer."""
+
+    @property
+    @abstractmethod
+    def catalog_context(self) -> CatalogSelection: ...
+
+    async def prepare_field(self, item: TargetField, catalogs: ConfigOptions) -> TargetField:
+        return await item.prepare_catalog(catalogs, self.catalog_context)
+
+
+@dataclass(frozen=True)
+class CommandContext(CatalogSelection):
+    """A declaration of supplied fields, not an alternate command editor; the one declared
+    command inheriting it owns the fields it lacks."""
+
+    @classmethod
+    def command(cls) -> type[CliCommand]:
+        (command,) = CliCommand.members_with(cls)
+        return command
+
+
+@dataclass(frozen=True)
+class PendingCommandBinding(CatalogBinding):
+    """Only supplied context, borrowing the concrete command's declared fields."""
+
+    context: CommandContext
+
+    @property
+    def declaration(self) -> type[CliCommand]:
+        return type(self.context).command()
+
+    @property
+    def catalog_context(self) -> CatalogSelection:
+        return self.context
+
+    def retained_payload(self) -> dict[str, object]:
+        return {entry.name: entry.current(self.context)
+                for entry in FieldCodec.field_plans(type(self.context))}
+
+    def field_value(self, entry: FieldPlan) -> TargetField:
+        for supplied in FieldCodec.field_plans(type(self.context)):
+            if supplied.declared is entry.declared:
+                return TargetField.of(entry, supplied.current(self.context))
+        return TargetField.omitted(entry)
+
+    def for_fields(self, arguments: Mapping[str, object]) -> PendingCommandBinding:
+        self.authorize_edits(arguments)
+        context_fields = {entry.key for entry in FieldCodec.argument_fields(type(self.context))}
+        wire = FieldCodec.editor_arguments(type(self.context), {
+            key: value for key, value in arguments.items() if key in context_fields
+        })
+        changes = self.decoded_edits(type(self.context), wire)
+        return replace(self, context=replace(self.context, **changes)) if changes else self
+
+    def confirmation(self) -> str:
+        return ''
+
+    @classmethod
+    def selection_confirmation(cls, commands: tuple[Self, ...]) -> str:
+        return ''
+
+    def with_confirmation(self, confirmed: bool) -> CliCommand:
+        return self.edited({}).with_confirmation(confirmed)
+
+
 @dataclass(frozen=True, kw_only=True)
 class CliCommand(
     ModuleMembers,
     DeclaredCommand,
     CliExecution[Answer],
+    CommandBinding,
     Generic[Answer],
     affix="CliCommand",
     discovery_package="agent_comms.cli_verbs",
 ):
+    @property
+    def declaration(self) -> type[Self]:
+        return type(self)
+
     def schedule(self, batch: TargetBatchSchedule, comms: Comms) -> None:
         result = TargetCompleted(batch.target, self, self.apply(comms))
         batch.outcomes.append((batch.position, result))
@@ -125,23 +268,47 @@ def selection_channels(cls, channel: ChannelSelection,
         return channel.channels_for(target)
 
     @classmethod
-    def thread_bindings(cls, comms: Comms, thread: Thread, status: ThreadStatus, channel: SurroundingChannel, *, catalog: CatalogAccess) -> tuple[Self, ...]:
+    def thread_bindings(cls, comms: Comms, thread: Thread, status: ThreadStatus, channel: SurroundingChannel, *, catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
         return ()
 
     @classmethod
-    def project_bindings(cls, comms: Comms, project: Checkout) -> tuple[Self, ...]:
+    def project_bindings(cls, comms: Comms, project: Checkout) -> tuple[CommandBinding, ...]:
         return ()
 
     @classmethod
-    def channel_bindings(cls, comms: Comms, resolved: ResolvedChannel, *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess) -> tuple[Self, ...]:
+    def channel_bindings(cls, comms: Comms, resolved: ResolvedChannel, *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
         return ()
 
+    @singledispatchmethod
     @classmethod
-    def bindings(cls, comms: Comms, target: CommandTarget, channel: SurroundingChannel,
-                 *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess,
-                 resolved: ResolvedThread | ResolvedChannel | Checkout) -> tuple[Self, ...]:
-        return target.bindings(
-            cls, comms, channel, resolved, project=project, snapshot=snapshot, catalog=catalog)
+    def bindings(cls, resolved: object, comms: Comms, channel: SurroundingChannel,
+                 *, project: Checkout, snapshot: RegistrySnapshot,
+                 catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
+        """The resolved target's type selects which declared bindings answer it."""
+        raise TypeError(f"{type(resolved).__qualname__} is not a resolved command target")
+
+    @bindings.register
+    @classmethod
+    def _thread_bindings(cls, resolved: ResolvedThread, comms: Comms, channel: SurroundingChannel,
+                         *, project: Checkout, snapshot: RegistrySnapshot,
+                         catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
+        return cls.thread_bindings(
+            comms, resolved.thread, resolved.status, channel, catalog=catalog)
+
+    @bindings.register
+    @classmethod
+    def _channel_bindings(cls, resolved: ResolvedChannel, comms: Comms, channel: SurroundingChannel,
+                          *, project: Checkout, snapshot: RegistrySnapshot,
+                          catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
+        return cls.channel_bindings(
+            comms, resolved, project=project, snapshot=snapshot, catalog=catalog)
+
+    @bindings.register
+    @classmethod
+    def _project_bindings(cls, resolved: Checkout, comms: Comms, channel: SurroundingChannel,
+                          *, project: Checkout, snapshot: RegistrySnapshot,
+                          catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
+        return cls.project_bindings(comms, resolved)
 
     @classmethod
     def target_catalog(cls, comms: Comms, target: TargetSelection,
@@ -160,13 +327,14 @@ def target_catalog(cls, comms: Comms, target: TargetSelection,
         catalog = (comms.channels.catalog.read()
                    if channel.needs_catalog or any(item.needs_channel_state for item in selected) else NoCatalog())
         resolved_targets = tuple((name, name.resolve(snapshot, comms.bus.require_channel)) for name in available_targets)
-        grouped: dict[type[CliCommand], dict[CliCommand, None]] = {}
+        grouped: dict[type[CliCommand], dict[CommandBinding, None]] = {}
         for member in target.command_members(cls):
             selected_channels = channel if member.channel_fields() else NoChannel()
             for name, resolved in resolved_targets:
                 for context in name.contexts(member, selected_channels):
                     available = member.bindings(
-                        comms, name, context, project=project, snapshot=snapshot, catalog=catalog, resolved=resolved)
+                        resolved, comms, context,
+                        project=project, snapshot=snapshot, catalog=catalog)
                     if available:
                         declaration = target.listed_declaration(member)
                         grouped.setdefault(declaration, {}).update(dict.fromkeys(
@@ -176,23 +344,13 @@ def target_catalog(cls, comms: Comms, target: TargetSelection,
                                   channel if declaration.channel_fields() else NoChannel())
                      for declaration, commands in grouped.items())
 
-    def for_editor(self, comms: Comms, target: str, project: Checkout) -> Self:
-        return self
-
-    @property
-    def editable_fields(self) -> tuple[TargetField, ...]:
-        """The command owns the parameters offered by its target action."""
-
-        return tuple(
-            TargetField.of(entry, entry.current(self))
-            for entry in FieldCodec.argument_fields(type(self))
-            if not entry.setting(TargetBound, False)
-        )
+    def field_value(self, entry: FieldPlan) -> TargetField:
+        return TargetField.of(entry, entry.current(self))
 
-    async def prepare_fields(self) -> tuple[TargetField, ...]:
-
-        catalogs = ConfigOptions.configured()
-        return tuple([await item.prepare(catalogs, self) for item in self.editable_fields])
+    def retained_payload(self) -> dict[str, object]:
+        return {entry.name: entry.current(self)
+                for entry in FieldCodec.argument_fields(type(self))
+                if entry.setting(TargetBound, False)}
 
     def describe(self, comms: Comms, target: CommandTarget, project: Checkout) -> TargetAction:
 
@@ -235,23 +393,6 @@ def with_confirmation(self, confirmed: bool) -> Self:
             raise ValueError(self.confirmation())
         return self
 
-    def edited(self, arguments: Mapping[str, object]) -> Self:
-        bound = {
-            entry.key for entry in FieldCodec.argument_fields(type(self))
-            if entry.setting(TargetBound, False)
-        }
-        if bound & arguments.keys():
-            raise ValueError("Target-bound parameters cannot be overridden")
-        editable = {item.name for item in self.editable_fields}
-        if arguments.keys() - editable:
-            raise ValueError("Parameters not offered by this target command cannot be overridden")
-        encoded = FieldCodec.encode(self)
-        return type(self).from_payload({
-            **{key: encoded[key] for key in bound & encoded.keys()},
-            type(self).family_discriminator: FieldCodec.encode(type(self)),
-            **arguments,
-        })
-
     @classmethod
     def execute_target(cls, comms: Comms, target: TargetSelection, arguments: Mapping[str, object],
                        *, confirmed: bool = False,
@@ -271,7 +412,7 @@ def execute_target(cls, comms: Comms, target: TargetSelection, arguments: Mappin
 @dataclass(frozen=True)
 class TargetAction:
     """A selection keeps the bound commands, not a separate permission list."""
-    bound: tuple[CliCommand, ...]
+    bound: tuple[CommandBinding, ...]
     targets: tuple[str, ...]
     declaration: type[CliCommand]
     channels: ChannelSelection = NoChannel()
@@ -280,8 +421,10 @@ class TargetAction:
     def editable_fields(self) -> tuple[TargetField, ...]:
         return self.bound[0].editable_fields
 
-    async def prepare_fields(self) -> tuple[TargetField, ...]:
-        return await self.bound[0].prepare_fields()
+    async def prepare_fields(
+        self, arguments: Mapping[str, object] = MappingProxyType({}),
+    ) -> tuple[TargetField, ...]:
+        return await self.bound[0].prepare_fields(arguments)
 
     @property
     def label(self) -> str:
@@ -289,14 +432,14 @@ def label(self) -> str:
 
     @property
     def confirmation(self) -> str:
-        groups: dict[type[CliCommand], list[CliCommand]] = {}
+        groups: dict[type[CommandBinding], list[CommandBinding]] = {}
         for command in self.bound:
             groups.setdefault(type(command), []).append(command)
         return '\n'.join(filter(None, (declaration.selection_confirmation(tuple(commands))
                                        for declaration, commands in groups.items())))
 
     def edited(self, arguments: Mapping[str, object]) -> TargetAction:
-        return replace(self, bound=tuple(command.edited(type(command).editor_arguments(arguments))
+        return replace(self, bound=tuple(command.edited(command.declaration.editor_arguments(arguments))
                                         for command in self.bound))
 
     def with_confirmation(self, confirmed: bool) -> TargetAction:
@@ -442,8 +585,9 @@ def run(self, declaration: type[CliCommand], comms: Comms, arguments: Mapping[st
         name, = self.targets
         resolved = name.resolve(snapshot, comms.bus.require_channel)
         bindings = tuple(bound for context in name.contexts(declaration, channel)
-                         for bound in declaration.bindings(comms, name, context, project=project, snapshot=snapshot,
-                                                           catalog=catalog, resolved=resolved))
+                         for bound in declaration.bindings(
+                             resolved, comms, context,
+                             project=project, snapshot=snapshot, catalog=catalog))
         if len(bindings) == 1:
             bound = bindings[0].for_editor(comms, name.name, project).edited(arguments)
             return bound.with_confirmation(confirmed).apply(comms)
@@ -480,7 +624,7 @@ def run(self, declaration: type[CliCommand], comms: Comms, arguments: Mapping[st
                               if member.catalog_declaration() is declaration)
                         if declaration.catalog_declaration() is declaration else (declaration,))
         outcomes: list[tuple[int, TargetCompleted | TargetFailed]] = []
-        planned: dict[CliCommand, tuple[int, CommandTarget, SurroundingChannel]] = {}
+        planned: dict[CommandBinding, tuple[int, CommandTarget, SurroundingChannel]] = {}
         together: dict[type[AppliedTogether], list[tuple[int, CommandTarget, AppliedTogether]]] = {}
         for name in self.targets:
             try:
@@ -492,8 +636,9 @@ def run(self, declaration: type[CliCommand], comms: Comms, arguments: Mapping[st
             for context in name.contexts(declaration, channel):
                 try:
                     bindings = tuple(command for member in declarations
-                                     for command in member.bindings(comms, name, context,
-                                                                   project=project, snapshot=snapshot, catalog=catalog, resolved=resolved))
+                                     for command in member.bindings(
+                                         resolved, comms, context,
+                                         project=project, snapshot=snapshot, catalog=catalog))
                     if not bindings:
                         raise CommandSelectionChanged(
                             f'This action is no longer available for {name}{context.scope()}')
@@ -511,9 +656,9 @@ def run(self, declaration: type[CliCommand], comms: Comms, arguments: Mapping[st
                 current = comms.registry.snapshot()
                 current_catalog = (comms.channels.catalog.read()
                                    if context.needs_catalog or name.needs_channel_state else NoCatalog())
-                if bound_command not in type(bound_command).bindings(
-                        comms, name, context, project=project, snapshot=current, catalog=current_catalog,
-                        resolved=name.resolve(current, comms.bus.require_channel)):
+                if bound_command not in bound_command.declaration.bindings(
+                        name.resolve(current, comms.bus.require_channel), comms, context,
+                        project=project, snapshot=current, catalog=current_catalog):
                     raise CommandSelectionChanged(
                         f'The bound command for {name}{context.scope()} changed; refresh the selection')
                 command.schedule(TargetBatchSchedule(position, name, outcomes, together), comms)
diff --git a/src/agent_comms/cli_commands.py b/src/agent_comms/cli_commands.py
index 81956803eb..52597279d0 100644
--- a/src/agent_comms/cli_commands.py
+++ b/src/agent_comms/cli_commands.py
@@ -66,16 +66,6 @@ def _tags(value: str) -> list[str]:
     return [tag.strip() for tag in value.split(",") if tag.strip()]
 
 
-@boundary_decoder
-def _tag_choice_text(value: str | None) -> str | None:
-    """Command-line and editor text is the given tags' wire; no text inherits."""
-    return value
-
-
-def _tag_text(value: frozenset[str] | None) -> str | None:
-    return None if value is None else ', '.join(sorted(value))
-
-
 def _shell_words(value: str) -> list[str]:
     return shlex.split(value)
 
diff --git a/src/agent_comms/cli_support/named_thread.py b/src/agent_comms/cli_support/named_thread.py
index badadaaecd..be9f7808d8 100644
--- a/src/agent_comms/cli_support/named_thread.py
+++ b/src/agent_comms/cli_support/named_thread.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from agent_comms.catalog_document import CatalogAccess
+from agent_comms.cli_command import CommandBinding
 from agent_comms.channel_targets import OneChannel, ResolvedChannel
 from agent_comms.channel_targets import ResolvedThread
 from agent_comms.cli_support.named import NamedCliCommand
@@ -24,7 +25,7 @@ def channel_members(cls, resolved: ResolvedChannel) -> tuple[ResolvedThread, ...
         return ()
 
     @classmethod
-    def channel_bindings(cls, comms: Comms, resolved: ResolvedChannel, *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess) -> tuple[Self, ...]:
+    def channel_bindings(cls, comms: Comms, resolved: ResolvedChannel, *, project: Checkout, snapshot: RegistrySnapshot, catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
         return tuple(bound for member in cls.channel_members(resolved)
                      for bound in cls.thread_bindings(
                          comms, member.thread, member.status, OneChannel(resolved.channel.name), catalog=catalog))
diff --git a/src/agent_comms/cli_support/thread_bound.py b/src/agent_comms/cli_support/thread_bound.py
index cac53354a5..6b6e2d8129 100644
--- a/src/agent_comms/cli_support/thread_bound.py
+++ b/src/agent_comms/cli_support/thread_bound.py
@@ -4,13 +4,12 @@
 
 from abc import abstractmethod
 from agent_comms.catalog_document import CatalogAccess
-from agent_comms.cli_command import CliCommand
+from agent_comms.cli_command import CliCommand, CommandBinding
 from agent_comms.comms import Comms
 from agent_comms.thread_execution import ThreadCommandAvailability
 from agent_comms.thread_status import ThreadStatus
 from agent_comms.threads import Thread
 from dataclasses import dataclass
-from typing import Self
 
 
 @dataclass(frozen=True, kw_only=True)
@@ -18,9 +17,9 @@ class ThreadBoundCliCommand(ThreadCommandAvailability, CliCommand):
     @classmethod
     def thread_bindings(cls, comms: Comms, thread: Thread, status: ThreadStatus,
                         channel: SurroundingChannel, *,
-                        catalog: CatalogAccess) -> tuple[Self, ...]:
+                        catalog: CatalogAccess) -> tuple[CommandBinding, ...]:
         return (cls.thread_binding(thread),) if cls.available_for_thread(thread, status) else ()
 
     @classmethod
     @abstractmethod
-    def thread_binding(cls, thread: Thread) -> Self: ...
+    def thread_binding(cls, thread: Thread) -> CommandBinding: ...
diff --git a/src/agent_comms/cli_support/thread_creation.py b/src/agent_comms/cli_support/thread_creation.py
new file mode 100644
index 0000000000..92bb80bab2
--- /dev/null
+++ b/src/agent_comms/cli_support/thread_creation.py
@@ -0,0 +1,61 @@
+"""The launch settings and field-derived construction shared by new and fork."""
+from __future__ import annotations
+
+import asyncio
+from abc import abstractmethod
+from dataclasses import dataclass, fields
+from pathlib import Path
+from typing import Annotated
+
+from agent_comms.cli_command import CatalogBinding, CliCommand, CommandContext
+from agent_comms.cli_commands import NameArgument, ThreadCreatedResult
+from agent_comms.comms import Comms
+from agent_comms.config_options import ConfigOptions, ModelConfigOption, ThinkingLevelConfigOption
+from agent_comms.nominal.field_codec import Argument, Catalog, Multiline
+from agent_comms.native_settings import NativeThreadDefaults
+from agent_comms.pi_vocabulary import ThinkingLevelField
+from agent_comms.repo import Checkout, LaunchCheckout
+from agent_comms.thread_management import ThreadCreationSpec
+from agent_comms.thread_model import CatalogSelection, ThreadModel
+
+
+@dataclass(frozen=True, kw_only=True)
+class ThreadLaunchSettings(CommandContext):
+    """Acquired launch settings; the creator has not supplied a child name."""
+    task: Annotated[str, Multiline(), Argument(flags=("--task",))] = ""
+    project: Annotated[Checkout, Argument(flags=("--project",))]
+    model: Annotated[
+        ThreadModel, Catalog(ModelConfigOption), Argument(flags=("--model",)),
+    ] = NativeThreadDefaults.model
+    thinking_level: Annotated[
+        ThinkingLevelField, Catalog(ThinkingLevelConfigOption), Argument(flags=("--thinking-level",)),
+    ] = NativeThreadDefaults.thinking_level
+
+    @property
+    def project_path(self) -> Path:
+        return self.project.path
+
+
+@dataclass(frozen=True, kw_only=True)
+class ThreadCreationCliCommand(
+    NameArgument, ThreadLaunchSettings, CatalogBinding, CliCommand[ThreadCreatedResult], declared_name=None,
+):
+    @property
+    @abstractmethod
+    def spec_type(self) -> type[ThreadCreationSpec]: ...
+
+    @property
+    def catalog_context(self) -> CatalogSelection:
+        return self
+
+    def specification(self) -> ThreadCreationSpec:
+        return self.spec_type.from_task(**{
+            **{item.name: getattr(self, item.name) for item in fields(self.spec_type)},
+            'project': LaunchCheckout.checked(self.project.path),
+        })
+
+    def apply(self, ctx: Comms) -> ThreadCreatedResult:
+        spec = self.specification()
+        asyncio.run(ConfigOptions.configured().validate(self))
+        child = ctx.threads.create(spec)
+        return ThreadCreatedResult(child.name, child.process_identity)
diff --git a/src/agent_comms/cli_verbs/fork.py b/src/agent_comms/cli_verbs/fork.py
index 199539f054..90426b441a 100644
--- a/src/agent_comms/cli_verbs/fork.py
+++ b/src/agent_comms/cli_verbs/fork.py
@@ -1,34 +1,32 @@
 from __future__ import annotations
 
-from agent_comms.channel_targets import DeclaredTag
-from agent_comms.cli_verbs.new import NewThreadCliCommand
+from agent_comms.cli_command import PendingCommandBinding
+from agent_comms.cli_support.thread_creation import ThreadCreationCliCommand, ThreadLaunchSettings
 from agent_comms.cli_support.thread_bound import ThreadBoundCliCommand
-from agent_comms.comms import Comms
 from agent_comms.nominal.field_codec import Argument
 from agent_comms.nominal.field_codec import TargetBound
 from agent_comms.thread_execution import NativeOwnerCommand
 from agent_comms.thread_management import ForkSpec
-from agent_comms.thread_management import GivenTags
+from agent_comms.thread_management import InheritedTags, TagChoice
 from agent_comms.threads import Thread
-from agent_comms.repo import Checkout
 from dataclasses import dataclass
 from typing import Annotated
-from typing import Self
 
 
 @dataclass(frozen=True, kw_only=True)
-class ForkCliCommand(NativeOwnerCommand, ThreadBoundCliCommand, NewThreadCliCommand):
+class ForkContext(ThreadLaunchSettings):
+    parent: Annotated[str, TargetBound(), Argument(flags=("--parent",))]
+
+
+@dataclass(frozen=True, kw_only=True)
+class ForkCliCommand(NativeOwnerCommand, ThreadBoundCliCommand, ForkContext, ThreadCreationCliCommand):
     help = "Fork a child pi thread"
     spec_type = ForkSpec
-    parent: Annotated[
-        str, TargetBound(), Argument(flags=("--parent",))
-    ]
 
-    @classmethod
-    def project_bindings(cls, comms: Comms, project: Checkout) -> tuple[Self, ...]:
-        return ()
+    tags: Annotated[TagChoice, Argument()] = InheritedTags()
 
     @classmethod
-    def thread_binding(cls, thread: Thread) -> Self:
-        return cls(name='', parent=thread.name, tags=GivenTags(DeclaredTag.for_child(thread.tags)), project=thread.worktree,
-                   model=thread.model, thinking_level=thread.thinking_level)
+    def thread_binding(cls, thread: Thread) -> PendingCommandBinding:
+        return PendingCommandBinding(ForkContext(
+            parent=thread.name, project=thread.worktree,
+            model=thread.model, thinking_level=thread.thinking_level))
diff --git a/src/agent_comms/cli_verbs/new.py b/src/agent_comms/cli_verbs/new.py
index 7b0e803019..1d95ce53fe 100644
--- a/src/agent_comms/cli_verbs/new.py
+++ b/src/agent_comms/cli_verbs/new.py
@@ -1,76 +1,31 @@
 from __future__ import annotations
 
-from agent_comms.cli_command import CliCommand
-from agent_comms.cli_commands import NameArgument
-from agent_comms.cli_commands import ThreadCreatedResult
-from agent_comms.cli_commands import _tag_choice_text
+from agent_comms.cli_command import PendingCommandBinding
+from agent_comms.cli_support.thread_creation import ThreadCreationCliCommand, ThreadLaunchSettings
 from agent_comms.comms import Comms
-from agent_comms.config_options import ConfigOptions
-from agent_comms.config_options import ModelConfigOption
-from agent_comms.config_options import ThinkingLevelConfigOption
 from agent_comms.nominal.field_codec import Argument
-from agent_comms.nominal.field_codec import Catalog
-from agent_comms.nominal.field_codec import EditorFormat
-from agent_comms.nominal.field_codec import FieldCodec
-from agent_comms.nominal.field_codec import Multiline
-from agent_comms.nominal.field_codec import Normalize
 from agent_comms.native_settings import NativeThreadDefaults
-from agent_comms.repo import Checkout, LaunchCheckout
-from agent_comms.pi_vocabulary import ThinkingLevelField
-from agent_comms.thread_management import InheritedTags
+from agent_comms.repo import Checkout
+from agent_comms.thread_management import GivenTags
 from agent_comms.thread_management import NewThreadSpec
-from agent_comms.thread_management import TagChoice
-from agent_comms.thread_model import CatalogSelection, ThreadModel
-from pathlib import Path
 from dataclasses import dataclass
-from dataclasses import fields
-from typing import Annotated
-from typing import ClassVar
-from typing import Self
-import asyncio
+from typing import Annotated, ClassVar
 
 
 @dataclass(frozen=True, kw_only=True)
-class NewThreadCliCommand(NameArgument, CatalogSelection, CliCommand, declared_name="new"):
+class NewThreadContext(ThreadLaunchSettings):
+    """The launch settings a project offers before the creator names the child."""
+
+
+@dataclass(frozen=True, kw_only=True)
+class NewThreadCliCommand(NewThreadContext, ThreadCreationCliCommand, declared_name="new"):
     help = "Create a new pi thread"
     spec_type: ClassVar[type[NewThreadSpec]] = NewThreadSpec
-    task: Annotated[
-        str, Multiline(), Argument(flags=("--task",))
-    ] = ""
-    tags: Annotated[
-        TagChoice,
-        EditorFormat(FieldCodec.encode),
-        Normalize(_tag_choice_text),
-        Argument(flags=("--tags",)),
-    ] = InheritedTags()
-    project: Annotated[
-        Checkout,
-        Argument(flags=("--project",)),
-    ]
-    model: Annotated[
-        ThreadModel, Catalog(ModelConfigOption), Argument(flags=("--model",)),
-    ] = NativeThreadDefaults.model
-    thinking_level: Annotated[
-        ThinkingLevelField,
-        Catalog(ThinkingLevelConfigOption),
-        Argument(flags=("--thinking-level",)),
-    ] = NativeThreadDefaults.thinking_level
+    tags: Annotated[GivenTags, Argument()]
 
     @classmethod
-    def project_bindings(cls, comms: Comms, project: Checkout) -> tuple[Self, ...]:
+    def project_bindings(cls, comms: Comms, project: Checkout) -> tuple[PendingCommandBinding, ...]:
         defaults = NativeThreadDefaults()
-        return (cls(name='', project=project, model=defaults.model,
-                    thinking_level=defaults.thinking_level),)
-
-    @property
-    def project_path(self) -> Path:
-        return self.project.path
-
-    def apply(self, ctx: Comms) -> ThreadCreatedResult:
-        spec = self.spec_type.from_task(**{
-            **{item.name: getattr(self, item.name) for item in fields(self.spec_type)},
-            'project': LaunchCheckout.checked(self.project.path),
-        })
-        asyncio.run(ConfigOptions.configured().validate(self))
-        child = ctx.threads.create(spec)
-        return ThreadCreatedResult(child.name, child.process_identity)
+        return (PendingCommandBinding(NewThreadContext(
+            project=project, model=defaults.model,
+            thinking_level=defaults.thinking_level)),)
diff --git a/src/agent_comms/cli_verbs/thread_tags.py b/src/agent_comms/cli_verbs/thread_tags.py
index 5ee2beeb37..1bcdf51ab6 100644
--- a/src/agent_comms/cli_verbs/thread_tags.py
+++ b/src/agent_comms/cli_verbs/thread_tags.py
@@ -4,13 +4,10 @@
 
 from agent_comms.catalog_document import CatalogAccess
 from agent_comms.cli_commands import ThreadTagsResult
-from agent_comms.cli_commands import _tag_text
-from agent_comms.cli_commands import _tags
 from agent_comms.cli_support.named import NamedCliCommand
 from agent_comms.comms import Comms
 from agent_comms.nominal.field_codec import Argument
-from agent_comms.nominal.field_codec import EditorFormat
-from agent_comms.nominal.field_codec import Normalize
+from agent_comms.launch_runtime import TagListText
 from agent_comms.thread_status import ThreadStatus
 from agent_comms.threads import Thread
 from dataclasses import dataclass
@@ -23,8 +20,7 @@ class ThreadTagsCliCommand(NamedCliCommand, declared_name="thread-tags"):
     help = "Edit thread tags"
     tags: Annotated[
         frozenset[str],
-        Normalize(_tags),
-        EditorFormat(_tag_text),
+        TagListText,
         Argument(
             flags=("--tags",), options={"help": "Complete tags, separated by commas"}
         ),
diff --git a/src/agent_comms/nominal/field_codec.py b/src/agent_comms/nominal/field_codec.py
index 639277d501..cf3d4d0207 100644
--- a/src/agent_comms/nominal/field_codec.py
+++ b/src/agent_comms/nominal/field_codec.py
@@ -14,7 +14,7 @@
 import types
 from abc import ABC, abstractmethod
 from collections.abc import Callable, Collection, Hashable, Iterable, Iterator, Mapping, Sequence
-from dataclasses import MISSING, Field, dataclass, field, fields, is_dataclass
+from dataclasses import MISSING, Field, dataclass, field, fields, is_dataclass, make_dataclass
 from datetime import datetime
 from enum import Enum, IntFlag
 from functools import cache, cached_property, lru_cache, partial, reduce
@@ -394,10 +394,6 @@ class Normalize(ValueDirective[Callable[[object], object]]):
     pass
 
 
-class EditorFormat(ValueDirective[Callable[[object], JsonValue]]):
-    pass
-
-
 class Editor(ValueDirective[type]):
     pass
 
@@ -679,18 +675,15 @@ def argument_normalizer(self) -> Callable[[object], object]:
     def editor_normalizer(self) -> Callable[[object], object]:
         return self.setting(Normalize, self.value.decode_editor)
 
-    @cached_property
-    def editor_formatter(self) -> Callable[[object], JsonValue]:
-        return self.setting(EditorFormat, self.value.encode_editor)
-
     def argument_payload(self, value: object) -> object:
         return self.argument_normalizer(value)
 
     def editor_payload(self, value: object) -> object:
         return self.editor_normalizer(value)
 
-    def editor_default(self, value: object) -> JsonValue:
-        return self.editor_formatter(value)
+    def require_constructor_input(self, values: Mapping[str, object]) -> None:
+        if self.constructor_required and self.name not in values:
+            raise MalformedRecord(f'Missing required field {self.key}')
 
     @property
     def description(self) -> str:
@@ -1033,6 +1026,44 @@ class DecoderPlan(ABC):
 
     target: object
 
+    @property
+    def editor_annotation(self) -> object:
+        return str
+
+    @property
+    def native_editor_annotation(self) -> object:
+        return self.editor_annotation
+
+    @property
+    def editor_members(self) -> tuple[EditorMember, ...]:
+        return ()
+
+    def editor_member(self, declared: DecoderPlan) -> EditorMember:
+        return ScalarEditorMember(declared)
+
+    def selected_editor_member(self, value: object) -> EditorMember:
+        return self.editor_member(self)
+
+    def project_editor(self, native: object, wire: Callable[[], object]) -> object:
+        return self.format_editor(wire())
+
+    def editor_argument(self, value: object) -> object:
+        return value
+
+    def accepts_value(self, value: object) -> bool:
+        """Select a native member at this decoder boundary."""
+        return isinstance(value, self.target)
+
+    @abstractmethod
+    def decode_bound(self, data: object, context: Mapping[str, object]) -> object: ...
+
+    def format_editor(self, wire: object) -> str:
+        return json.dumps(wire)
+
+    @boundary_decoder
+    def parse_editor(self, text: object) -> object:
+        return None if text is None else FieldCodec.parse(FieldCodec.decode(str, text))
+
     def sql_checks(self, column: str, quote: Callable[[object], str]) -> tuple[str, ...]:
         """Stored-value predicates; the SQL owner quotes its column and constants."""
         return ()
@@ -1064,6 +1095,123 @@ def external_decoders(self, data: object) -> Iterable[ExternalDecoder]:
         return ()
 
 
+class MemberDraftFactory[Draft](ABC):
+    """An integration constructs drafts; selections own which construction applies."""
+
+    @abstractmethod
+    def unchosen(self, member: UnchosenMember) -> Draft: ...
+
+    @abstractmethod
+    def member(self, member: EditorMember, value: object) -> Draft: ...
+
+
+class MemberSelection(ABC):
+    def argument(
+        self, key: str, value: object, decode: Callable[[object], object],
+    ) -> Mapping[str, object]:
+        return {key: decode(value)}
+
+    @property
+    @abstractmethod
+    def declaration(self) -> object: ...
+
+    @property
+    @abstractmethod
+    def label(self) -> str: ...
+
+    @abstractmethod
+    def draft[Draft](self, factory: MemberDraftFactory[Draft], value: object) -> Draft: ...
+
+
+@dataclass(frozen=True)
+class UnchosenMember(MemberSelection):
+    def argument(
+        self, key: str, value: object, decode: Callable[[object], object],
+    ) -> Mapping[str, object]:
+        return {}
+
+    @property
+    def declaration(self) -> object:
+        return type(self)
+
+    @property
+    def label(self) -> str:
+        return 'Choose a member…'
+
+    def draft[Draft](self, factory: MemberDraftFactory[Draft], value: object) -> Draft:
+        return factory.unchosen(self)
+
+
+@dataclass(frozen=True)
+class EditorMember(MemberSelection):
+    """Read-only member metadata; only actual inputs enter a form's ObjectState."""
+
+    decoder: DecoderPlan
+
+    @property
+    def declaration(self) -> object:
+        return self.decoder.target
+
+    @property
+    def label(self) -> str:
+        return str(self.declaration)
+
+    def draft[Draft](self, factory: MemberDraftFactory[Draft], value: object) -> Draft:
+        return factory.member(self, value)
+
+    @property
+    @abstractmethod
+    def fields(self) -> tuple[FieldPlan, ...]: ...
+
+    def current(self, entry: FieldPlan, value: object) -> object:
+        return value
+
+    @abstractmethod
+    def wire_payload(self, arguments: Mapping[str, object]) -> object: ...
+
+    def complete(self, arguments: Mapping[str, object]) -> object:
+        return self.decoder.decode_bound(self.wire_payload(arguments), {})
+
+
+class ScalarEditorMember(EditorMember):
+    @cached_property
+    def fields(self) -> tuple[FieldPlan, ...]:
+        declaration = make_dataclass('MemberInput', [('value', self.declaration)],
+                                     frozen=True, kw_only=True)
+        return FieldCodec.field_plans(declaration)
+
+    def wire_payload(self, arguments: Mapping[str, object]) -> object:
+        entry, = self.fields
+        return entry.editor_payload(entry.raw(dict(arguments)))
+
+
+@dataclass(frozen=True)
+class RecordEditorMember(EditorMember):
+    plans: tuple[FieldPlan, ...]
+
+    @property
+    def fields(self) -> tuple[FieldPlan, ...]:
+        return self.plans
+
+    def current(self, entry: FieldPlan, value: object) -> object:
+        return entry.current(value)
+
+    def wire_payload(self, arguments: Mapping[str, object]) -> object:
+        return {entry.key: entry.editor_payload(arguments[entry.key])
+                for entry in self.fields if entry.key in arguments}
+
+
+class NullEditorMember(EditorMember):
+    @property
+    def fields(self) -> tuple[FieldPlan, ...]:
+        return ()
+
+    def wire_payload(self, arguments: Mapping[str, object]) -> object:
+        if arguments:
+            raise MalformedRecord('A null member has no editor inputs')
+        return None
+
+
 @dataclass(frozen=True, slots=True)
 class ValueDecoder[T](DecoderPlan, JsonSchemaOwner):
     """One annotation owns its wire conversion, draft projection and choice domain."""
@@ -1129,10 +1277,9 @@ def representation_plan(
     @classmethod
     @cache
     def for_members[M](cls, members: tuple[type[M], ...]) -> UnionDecoder[M]:
-        return UnionDecoder(
-            reduce(or_, members),
-            tuple(compile_decoder(member) for member in members),
-        )
+        target = reduce(or_, members)
+        alternatives = tuple(compile_decoder(member) for member in members)
+        return UnionDecoder(target, alternatives)
 
     @abstractmethod
     def decode(self, data: object) -> T:
@@ -1172,10 +1319,6 @@ def decode_resolved(self, value: T) -> T:
             raise MalformedRecord(f"Resolved value does not match {self.target}")
         return value
 
-    def accepts_value(self, value: object) -> bool:
-        """Select an in-process union case using its compiled declaration."""
-        return isinstance(value, self.target)
-
     def accepts_external_value(self, value: object) -> bool:
         """A structural container match does not validate its contents."""
         return False
@@ -1199,15 +1342,18 @@ def decode_selected(self, data, selections):
     def decode_keys(self, data, keys, selections):
         raise TypeError("Selected keys require a mapping field")
 
-    editor_annotation = str
+    @property
+    def editor_wire(self) -> DecoderPlan:
+        return self
+
+    def editor_initial(self, value: T) -> object:
+        return self.editor_wire.project_editor(value, lambda: self.encode(value))
 
     def encode_editor(self, value: T) -> str:
-        encoded = FieldCodec.encode(value, self.target)
-        return json.dumps(encoded)
+        return self.editor_wire.format_editor(self.encode(value))
 
-    @boundary_decoder
-    def decode_editor(self, value):
-        return None if value is None else json.loads(value)
+    def decode_editor(self, value: object) -> object:
+        return self.editor_wire.parse_editor(value)
 
     def editor_choices(self) -> tuple[tuple[str, object], ...]:
         raise TypeError("The declared decoder does not offer finite choices")
@@ -1242,6 +1388,10 @@ def sql_checks(self, column: str, quote: Callable[[object], str]) -> tuple[str,
     def wire_structure(self) -> DecoderPlan:
         return self.representation.wire_structure(self.structural)
 
+    @property
+    def editor_wire(self) -> DecoderPlan:
+        return self.wire_structure
+
     @property
     def accepts_null(self) -> bool:
         return self.wire_structure.accepts_null
@@ -2931,6 +3081,12 @@ def fits_wire(self, data: object) -> bool:
     def accepts_value(self, value: object) -> bool:
         return self.accepts(value)
 
+    def parse_editor(self, text: object) -> object:
+        value = super().parse_editor(text)
+        if not self.accepts_value(value):
+            raise MalformedRecord(f'Expected native {self.target}, received {type(value).__name__}')
+        return value
+
     def accepts_external_value(self, value: object) -> bool:
         return self.accepts(value)
 
@@ -2955,6 +3111,9 @@ class _NullDecoder(_ExactScalarDecoder[None]):
     annotations = (type(None),)
     json_type = "null"
 
+    def editor_member(self, declared: DecoderPlan) -> EditorMember:
+        return NullEditorMember(declared)
+
     def sql_checks(self, column: str, quote: Callable[[object], str]) -> tuple[str, ...]:
         return (f"{column} IS NULL",)
 
@@ -2967,11 +3126,11 @@ class _TextDecoder(_ExactScalarDecoder[str]):
     def encode(self, value: str) -> str:
         return str(super().encode(value))
 
-    def encode_editor(self, value: str) -> str:
-        return value
+    def format_editor(self, wire: object) -> str:
+        return self.decode(wire)
 
-    def decode_editor(self, value):
-        return value
+    def parse_editor(self, text: object) -> object:
+        return text
 
 
 class _FlagDecoder(_ExactScalarDecoder[bool]):
@@ -3199,13 +3358,31 @@ def decode_external(self, data: object, record: Mapping[str, object]) -> T:
     def editor_annotation(self):
         return Union[tuple(item.editor_annotation for item in self.alternatives)]
 
+    @property
+    def native_editor_annotation(self) -> object:
+        return self.target
+
+    @property
+    def editor_members(self) -> tuple[EditorMember, ...]:
+        return tuple(member.editor_wire.editor_member(member) for member in self.alternatives)
+
+    def selected_editor_member(self, value: object) -> EditorMember:
+        member = self.selected(value)
+        return member.editor_wire.editor_member(member)
+
+    def project_editor(self, native: object, wire: Callable[[], object]) -> object:
+        return native
+
+    def editor_argument(self, value: object) -> object:
+        return self.format_editor(FieldCodec.encode(value, self.target))
+
     def argument_options(self, argument: Argument) -> dict[str, object]:
         return self.alternatives[0].argument_options(argument)
 
     def accepts_value(self, value: object) -> bool:
         return any(alternative.accepts_value(value) for alternative in self.alternatives)
 
-    def selected(self, value: T) -> ValueDecoder[T]:
+    def selected(self, value: object) -> ValueDecoder[T]:
         for alternative in self.alternatives:
             if alternative.accepts_value(value):
                 return alternative
@@ -3596,6 +3773,9 @@ class _RecordDecoder[T](_ContextDecoder[T]):
     def plan(self) -> FieldCodec._RecordPlan:
         return FieldCodec._record_plan(self.target)
 
+    def editor_member(self, declared: DecoderPlan) -> EditorMember:
+        return RecordEditorMember(declared, self.plan.bindings)
+
     def fits_wire(self, data: object) -> TypeGuard[dict[str, object]]:
         if not isinstance(data, dict):
             return False
diff --git a/src/agent_comms/target_fields.py b/src/agent_comms/target_fields.py
index a5c67041a5..7f2622d603 100644
--- a/src/agent_comms/target_fields.py
+++ b/src/agent_comms/target_fields.py
@@ -1,37 +1,47 @@
 from __future__ import annotations
 
+from abc import ABC, abstractmethod
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, replace
+from functools import singledispatchmethod
+
 from pydantic import JsonValue
-from typing import cast
 
 from agent_comms.nominal import field_codec
-from agent_comms.nominal.field_codec import FieldCodec
-from collections.abc import Callable
-from dataclasses import dataclass
-from dataclasses import replace
+from agent_comms.nominal.field_codec import FieldCodec, boundary_decoder
 from agent_comms.config_options import CatalogChoice, CatalogConfigOption, ConfigOptions
 from agent_comms.thread_model import CatalogSelection
 
 
 @dataclass(frozen=True)
-class TargetField:
-    """The declared field owns parsing, editor hints and requiredness."""
+class TargetField(ABC):
+    """The declared field owns parsing, editor hints and input presence."""
 
     plan: field_codec.FieldPlan
-    value: object
 
     @classmethod
     def of(cls, plan: field_codec.FieldPlan, value: object) -> TargetField:
-        editor: type[TargetField] = (
+        editor: type[SuppliedTargetField] = (
             CatalogTargetField
             if field_codec.Catalog.single(plan.annotation)
-            else plan.setting(field_codec.Editor, cls)
+            else plan.setting(field_codec.Editor, SuppliedTargetField)
         )
         return editor(plan, value)
 
+    @classmethod
+    def omitted(cls, plan: field_codec.FieldPlan) -> TargetField:
+        if plan.constructor_required:
+            return UnsuppliedTargetField(plan)
+        return cls.of(plan, plan.default())
+
     @property
     def annotation(self) -> object:
         return self.plan.annotation
 
+    @property
+    def editor_annotation(self) -> object:
+        return self.plan.value.editor_wire.native_editor_annotation
+
     @property
     def name(self) -> str:
         return self.plan.key
@@ -45,16 +55,40 @@ def description(self) -> str:
         return self.plan.description
 
     @property
-    def editor_default(self) -> JsonValue:
-        return self.plan.editor_default(self.value)
+    def editor_members(self) -> tuple[field_codec.EditorMember, ...]:
+        return self.plan.value.editor_wire.editor_members
+
+    @property
+    def editor_selections(self) -> tuple[field_codec.MemberSelection, ...]:
+        return self.editor_members
+
+    def editor_selection(self, value: object) -> field_codec.MemberSelection:
+        return self.plan.value.editor_wire.selected_editor_member(value)
+
+    @property
+    @abstractmethod
+    def editor_initial(self) -> object: ...
+
+    @abstractmethod
+    def default_schema(self) -> dict[str, JsonValue]: ...
+
+    def argument(self, value: object) -> Mapping[str, object]:
+        return {self.name: self.editor_argument(value)}
+
+    def editor_argument(self, value: object) -> object:
+        return self.plan.value.editor_wire.editor_argument(value)
 
     def editor_hints(self, choices: Callable[[ChoiceTargetField], object]) -> tuple[object, ...]:
         """An unrestricted field does not declare a finite-choice editor."""
         return ()
 
-    async def prepare(self, catalogs: ConfigOptions, selection: object) -> TargetField:
+    async def prepare(self, catalogs: ConfigOptions) -> TargetField:
         return self
 
+    async def prepare_catalog(self, catalogs: ConfigOptions,
+                              selection: CatalogSelection) -> TargetField:
+        return await self.prepare(catalogs)
+
     @property
     def required(self) -> bool:
         return self.plan.constructor_required
@@ -64,13 +98,78 @@ def value_schema(self) -> dict[str, JsonValue]:
 
     def json_schema(self) -> dict[str, JsonValue]:
         return {**self.value_schema(), 'multiline': self.multiline,
-                'description': self.description, 'editor_default': self.editor_default}
+                'description': self.description, **self.default_schema()}
+
 
 @dataclass(frozen=True)
-class ChoiceTargetField(TargetField):
+class SuppliedTargetField(TargetField):
+    value: object
+
+    @property
+    def editor_initial(self) -> object:
+        return self.plan.value.editor_initial(self.value)
+
+    @property
+    def editor_default(self) -> str:
+        return self.plan.value.encode_editor(self.value)
+
+    def default_schema(self) -> dict[str, JsonValue]:
+        return {'editor_default': self.editor_default}
+
+
+@dataclass(frozen=True)
+class UnsuppliedTargetField(TargetField):
+    """A required answer has no value; the optional editor owns its unset control."""
+
+    @property
+    def editor_annotation(self) -> object:
+        return self.plan.value.editor_wire.native_editor_annotation | type(None)
+
+    @property
+    def editor_initial(self) -> object:
+        return None
+
+    def default_schema(self) -> dict[str, JsonValue]:
+        return {}
+
+    @property
+    def editor_selections(self) -> tuple[field_codec.MemberSelection, ...]:
+        return (field_codec.UnchosenMember(), *super().editor_selections)
+
+    @singledispatchmethod
+    @boundary_decoder
+    def editor_selection(self, value: object) -> field_codec.MemberSelection:
+        return super().editor_selection(value)
+
+    @editor_selection.register
+    @boundary_decoder
+    def _unchosen_selection(self, value: None) -> field_codec.MemberSelection:
+        return field_codec.UnchosenMember()
+
+    def argument(self, value: object) -> Mapping[str, object]:
+        return self.editor_selection(value).argument(self.name, value, self.editor_argument)
+
+
+@dataclass(frozen=True)
+class ChoiceTargetField(SuppliedTargetField):
     """A prepared finite declaration, from its annotation or configuration catalog."""
     options: tuple[CatalogChoice, ...] = ()
 
+    @property
+    def editor_annotation(self) -> object:
+        return self.plan.value.editor_wire.editor_annotation
+
+    @property
+    def editor_members(self) -> tuple[field_codec.EditorMember, ...]:
+        return ()
+
+    @property
+    def editor_initial(self) -> object:
+        return self.editor_default
+
+    def editor_argument(self, value: object) -> object:
+        return value
+
     @property
     def editor_declarations(self) -> tuple[object, ...]:
         return ()
@@ -79,8 +178,7 @@ def editor_declarations(self) -> tuple[object, ...]:
     def choices(self) -> tuple[tuple[str, str], ...]:
         return tuple((choice.name, choice.editor_value) for choice in self.options)
 
-    async def prepare(self, catalogs: ConfigOptions,
-                      selection: object) -> ChoiceTargetField:
+    async def prepare(self, catalogs: ConfigOptions) -> ChoiceTargetField:
         decoder = field_codec.compile_decoder(self.annotation)
         options = (CatalogChoice(
                        name=label, value=decoder.encode_editor(decoder.decode(value)))
@@ -90,6 +188,7 @@ async def prepare(self, catalogs: ConfigOptions,
     def editor_hints(self, choices: Callable[[ChoiceTargetField], object]) -> tuple[object, ...]:
         return (choices(self),)
 
+
 class CatalogTargetField(ChoiceTargetField):
     @property
     def catalog(self) -> type[CatalogConfigOption]:
@@ -99,10 +198,10 @@ def catalog(self) -> type[CatalogConfigOption]:
     def editor_declarations(self) -> tuple[object, ...]:
         return (self.catalog,)
 
-    async def prepare(self, catalogs: ConfigOptions,
-                      selection: object) -> CatalogTargetField:
+    async def prepare_catalog(self, catalogs: ConfigOptions,
+                              selection: CatalogSelection) -> CatalogTargetField:
         catalog = catalogs.catalog_for(self.catalog)
-        options = await catalog.display_choices(cast(CatalogSelection, selection))
+        options = await catalog.display_choices(selection)
         decoder = field_codec.compile_decoder(self.annotation)
         return replace(self, options=tuple(CatalogChoice(name=item.name,
             value=decoder.encode_editor(decoder.decode(item.value))) for item in options))
diff --git a/src/agent_comms/thread_management.py b/src/agent_comms/thread_management.py
index 73cde4aa81..40e15d5eb9 100644
--- a/src/agent_comms/thread_management.py
+++ b/src/agent_comms/thread_management.py
@@ -4,7 +4,7 @@
 
 import asyncio
 import re
-from abc import abstractmethod
+from abc import ABC, abstractmethod
 from collections.abc import Sequence
 from contextlib import ExitStack
 from dataclasses import dataclass, field, replace
@@ -79,9 +79,6 @@ class RenameThreadResult(ToolResult):
 class TagChoice(SubclassUnion, affix="Tags"):
     """Select explicit, inherited or changed tags from an existing set."""
 
-    def for_new(self) -> frozenset[str]:
-        return self.for_child(frozenset())
-
     @abstractmethod
     def for_child(self, parent: frozenset[str]) -> frozenset[str]: ...
 
@@ -151,27 +148,36 @@ def local_directory(self) -> Path:
 
 
 @dataclass(frozen=True, slots=True, kw_only=True)
-class NewThreadSpec:
-    """Prepare the session, then register the thread under its registry lock."""
+class ThreadCreationSpec(ABC):
+    """Shared launch settings and construction after membership has been decided."""
 
     name: str
     project: LaunchCheckout
     model: ThreadModel
     thinking_level: ThinkingLevelField
     task: LaunchInput = field(default_factory=LaunchInput)
-    tags: TagChoice = InheritedTags()
 
     @classmethod
-    def from_task(cls, *, task: str = "", **values) -> NewThreadSpec:
+    def from_task(cls, *, task: str = "", **values) -> Self:
         return cls(task=TaskLaunchInput(task) if task else LaunchInput(), **values)
 
-    def declaration(self) -> Thread:
-        return Thread(name=self.name, task=self.task.thread_text(), tags=self.tags.for_new(),
+    def declaration(self, tags: frozenset[str]) -> Thread:
+        return Thread(name=self.name, task=self.task.thread_text(), tags=tags,
                       worktree=self.project, model=self.model, thinking_level=self.thinking_level)
 
+    @abstractmethod
+    def declare(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntime) -> Thread: ...
+
+
+@dataclass(frozen=True, slots=True, kw_only=True)
+class NewThreadSpec(ThreadCreationSpec):
+    """A new thread has no parent; its creator supplies its membership."""
+
+    tags: GivenTags
+
     def declare(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntime) -> Thread:
         project = self.project.path
-        thread = self.declaration()
+        thread = self.declaration(self.tags.tags)
         threads.registry.require_declaration_available(thread)
         location = thread.runtime.new_location(threads.root, project)
         saved = RegistrySessionLocation.from_path(location.read_key, threads.root)
@@ -179,10 +185,11 @@ def declare(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntime
 
 
 @dataclass(frozen=True, slots=True, kw_only=True)
-class ForkSpec(NewThreadSpec):
+class ForkSpec(ThreadCreationSpec):
     """A fork differs only in its parent-derived source and inherited settings."""
 
     parent: str
+    tags: TagChoice = InheritedTags()
 
     def _parent(self, threads: ThreadManagement) -> Thread:
         parent = threads.registry.require(self.parent)
@@ -192,8 +199,8 @@ def _parent(self, threads: ThreadManagement) -> Thread:
 
     def child(self, parent: Thread, runtime: RuntimeLocation, threads: ThreadManagement) -> Thread:
         saved = RegistrySessionLocation.from_path(runtime.read_key, threads.root)
-        return replace(self.declaration(),
-                       parent=ForkedParent(parent.name), tags=self.tags.for_child(DeclaredTag.for_child(parent.tags)),
+        return replace(self.declaration(self.tags.for_child(DeclaredTag.for_child(parent.tags))),
+                       parent=ForkedParent(parent.name),
                        backend=parent.backend, session_file=saved)
 
     def declare(self, threads: ThreadManagement, pi_bin: str, runtime: LaunchRuntime) -> Thread:
@@ -562,7 +569,7 @@ def _delete_originals(self, originals: Sequence[Thread]) -> tuple[ThreadDeleted,
                 for original in originals
             )
 
-    def create(self, spec: NewThreadSpec) -> Thread:
+    def create(self, spec: ThreadCreationSpec) -> Thread:
         """Prepare the session, then register and launch the thread with its task as first input."""
         resolved_bin = ConfiguredBinary().configured()
         runtime = LaunchRuntime.running()
diff --git a/tests/test_checkout_submodules.py b/tests/test_checkout_submodules.py
index 20ca488a3b..d136bf1718 100644
--- a/tests/test_checkout_submodules.py
+++ b/tests/test_checkout_submodules.py
@@ -137,7 +137,7 @@ def test_empty_project_does_not_become_the_callers_checkout(tmp_path, monkeypatc
     git(tmp_path, 'init', '-q')
     monkeypatch.chdir(tmp_path)
     comms = wire(tmp_path / 'wire')
-    result = core('--root', str(comms.root), 'new', '--name', 'empty-project', '--project', '')
+    result = core('--root', str(comms.root), 'new', '--name', 'empty-project', '--tags', '', '--project', '')
     assert result.returncode != 0
     assert 'Project path must not be empty' in result.stdout + result.stderr
     assert 'empty-project' not in comms.registry.snapshot().threads
diff --git a/tests/test_cli.py b/tests/test_cli.py
index 6197485483..9d674f4851 100644
--- a/tests/test_cli.py
+++ b/tests/test_cli.py
@@ -523,8 +523,8 @@ def test_channel_start_uses_native_visible_roster_without_launching(self, tmp_pa
         catalog = comms.channels.catalog.read()
         target = ChannelTarget("#team")
         commands = StartCliCommand.bindings(
-            comms, target, NoChannel(), project=Checkout(tmp_path), snapshot=snapshot, catalog=catalog,
-            resolved=target.resolve(snapshot, comms.bus.require_channel))
+            target.resolve(snapshot, comms.bus.require_channel), comms, NoChannel(),
+            project=Checkout(tmp_path), snapshot=snapshot, catalog=catalog)
         assert tuple(command.name for command in commands) == ("alpha",)
         assert all(comms.registry.require(name).process_identity == NoProcess()
                    for name in ("alpha", "external", "archived", "human"))
@@ -670,7 +670,10 @@ def test_result_kind_decides_exit_code_and_navigation_targets():
     assert not launched.open_targets() and not batch.open_targets()
     created = ThreadCreatedResult("child", NoProcess())
     assert created.open_targets() == ("child",)
-    command = NewThreadCliCommand(name="child", project=LaunchCheckout(Path("/project")), model=UndeclaredModel())
+    from agent_comms.thread_management import GivenTags
+
+    command = NewThreadCliCommand(name="child", tags=GivenTags(frozenset()),
+                                  project=LaunchCheckout(Path("/project")), model=UndeclaredModel())
     from agent_comms.channel_targets import ProjectTarget
 
     completed = TargetCompleted(ProjectTarget(Checkout(Path("/project"))), command, created)
diff --git a/tests/test_commander_notifications.py b/tests/test_commander_notifications.py
index fda2790c5b..8fab016a3c 100644
--- a/tests/test_commander_notifications.py
+++ b/tests/test_commander_notifications.py
@@ -15,8 +15,6 @@
 from agent_comms.registry_document import RegisteredCommanderProject
 
 from agent_comms.child_process import NoProcess, ProcessIdentity
-from agent_comms.cli_verbs.fork import ForkCliCommand
-from agent_comms.thread_management import GivenTags
 from agent_comms.commander_heartbeat import CommanderHeartbeat, CommanderObservation
 from agent_comms.comms import Comms
 from commander_heartbeat_fixture import ObservedHeartbeat, capture_frame, current_hud
@@ -962,17 +960,6 @@ def test_one_pass_gives_every_commander_its_own_packet(notification_bus):
         assert CommanderFrame.current(comms, recipient.name).packets == (quiet.task,)
 
 
-def test_fork_cli_default_inherits_audience_but_not_commander_role(notification_bus):
-    comms = notification_bus
-    parent = comms.registry.require('worker')
-    tag = CommandersTag.declared_name
-    parent = comms.registry.register(replace(parent, tags=parent.tags | {'comms', tag}))
-    command = ForkCliCommand.thread_binding(parent)
-    assert command.tags.for_new() == parent.tags - {tag}
-    explicit = replace(command, tags=GivenTags(parent.tags))
-    assert tag in explicit.tags.for_new()
-
-
 def test_a_commander_is_declared_never_inherited(notification_bus, tmp_path):
     """The hook's role answer reads the one derivation: a thread registered with the tag
     commands; a fork carries its commander's tags but acts for it, and a bound unit of a
diff --git a/tests/test_editor_defaults.py b/tests/test_editor_defaults.py
index e58f64e5d8..e0e9c05b53 100644
--- a/tests/test_editor_defaults.py
+++ b/tests/test_editor_defaults.py
@@ -4,7 +4,11 @@
 import pytest
 
 from agent_comms.cli_verbs.fork import ForkCliCommand
-from agent_comms.nominal.field_codec import FieldCodec, compile_decoder
+from dataclasses import dataclass
+
+from agent_comms.nominal.field_codec import FieldCodec, MalformedRecord, UnchosenMember, compile_decoder
+from agent_comms.target_fields import TargetField
+from agent_comms.thread_management import ChangedTags, GivenTags, InheritedTags, TagChoice
 from agent_comms.pi_vocabulary import HighThinkingLevel, NoThinkingOverride, ThinkingLevelField
 from agent_comms.repo import LaunchCheckout
 from agent_comms.thread_execution import ExternalThreadExecution, ThreadExecution
@@ -19,6 +23,10 @@
     (str | None, None, None),
     (bool, True, True),
     (tuple[int, ...], (1, 2), [1, 2]),
+    (TagChoice, GivenTags(frozenset({'auth'})), 'auth'),
+    (TagChoice, InheritedTags(), None),
+    (TagChoice, ChangedTags(frozenset({'auth'}), frozenset({'old'})),
+     {'add': ['auth'], 'remove': ['old']}),
 ))
 def test_declared_default_editor_round_trip(annotation, value, encoded):
     decoder = compile_decoder(annotation)
@@ -41,3 +49,74 @@ def test_fork_default_retains_annotated_thinking_representation(source_repositor
     assert compile_decoder(str).encode_editor('literal text') == 'literal text'
 
 
+@pytest.mark.parametrize('annotation,value', (
+    (TagChoice, GivenTags(frozenset())),
+    (TagChoice, GivenTags(frozenset({'auth'}))),
+    (TagChoice, InheritedTags()),
+    (TagChoice, ChangedTags(frozenset({'auth'}), frozenset({'old'}))),
+    (int | float, 3),
+    (int | float, 3.0),
+    (int | float | None, None),
+))
+def test_native_union_member_and_field_derived_transaction(annotation: object, value: object) -> None:
+    decoder = compile_decoder(annotation)
+    assert decoder.editor_initial(value) is value
+    member = decoder.editor_wire.selected_editor_member(value)
+    assert member.declaration is type(value)
+    assert tuple(item.declaration for item in decoder.editor_wire.editor_members).count(type(value)) == 1
+    parameters = tuple(TargetField.of(entry, member.current(entry, value)) for entry in member.fields)
+    arguments = {key: item for parameter in parameters
+                 for key, item in parameter.argument(parameter.editor_initial).items()}
+    result = member.complete(arguments)
+    assert type(result) is type(value)
+    assert result == value
+    json.dumps(decoder.encode_editor(value))
+    assert decoder.decode_editor(decoder.editor_wire.editor_argument(value)) == FieldCodec.encode(value, annotation)
+
+
+def test_overlapping_scalar_wire_does_not_change_selected_native_member() -> None:
+    decoder = compile_decoder(int | float)
+    member = decoder.editor_wire.selected_editor_member(3.0)
+    assert member.declaration is float
+    assert member.complete({'value': '3.0'}) == 3.0
+    with pytest.raises(MalformedRecord):
+        member.complete({'value': '3'})
+    with pytest.raises(MalformedRecord):
+        FieldCodec.decode(int | float, 3)
+
+
+def test_native_union_default_does_not_leak_into_public_json_schema() -> None:
+    plan = next(entry for entry in FieldCodec.argument_fields(ForkCliCommand) if entry.name == 'tags')
+    value = ChangedTags(frozenset({'auth'}), frozenset())
+    parameter = TargetField.of(plan, value)
+    assert parameter.editor_initial is value
+    assert parameter.editor_selection(value).declaration is ChangedTags
+    assert isinstance(parameter.editor_default, str)
+    json.dumps(parameter.json_schema())
+
+
+def test_unchosen_selection_is_not_a_supplied_null_member() -> None:
+    @dataclass(frozen=True)
+    class RequiredUnion:
+        value: int | float | None
+
+    plan, = FieldCodec.argument_fields(RequiredUnion)
+    omitted = TargetField.omitted(plan)
+    unchosen = omitted.editor_selection(omitted.editor_initial)
+    assert isinstance(unchosen, UnchosenMember)
+    assert unchosen.declaration is UnchosenMember
+    assert unchosen.label == 'Choose a member…'
+    assert unchosen.declaration in tuple(member.declaration for member in omitted.editor_selections)
+    assert omitted.editor_selection(3).declaration is int
+    supplied_null = TargetField.of(plan, None).editor_selection(None)
+    assert supplied_null.declaration is type(None)
+
+
+def test_record_member_inputs_remain_required_without_invented_values() -> None:
+    decoder = compile_decoder(TagChoice)
+    member = decoder.editor_wire.selected_editor_member(ChangedTags(frozenset(), frozenset()))
+    assert all(entry.constructor_required for entry in member.fields)
+    with pytest.raises(MalformedRecord):
+        member.complete({})
+
+
diff --git a/tests/test_end_to_end.py b/tests/test_end_to_end.py
index 9b41eb166a..12af3959cd 100644
--- a/tests/test_end_to_end.py
+++ b/tests/test_end_to_end.py
@@ -42,19 +42,21 @@ async def cli(programs, root, *arguments):
 
 
 class TestEndToEndLifecycle:
-    @pytest.mark.parametrize('operation,parent_arguments', [
-        ('new', ()), ('fork', ('--parent', 'parent')),
+    @pytest.mark.parametrize('operation,creation_arguments,expected_tags', [
+        ('new', ('--tags', 'chosen'), frozenset({'chosen'})),
+        ('new', ('--tags', ''), frozenset()),
+        ('fork', ('--parent', 'parent'), frozenset({'inherited'})),
     ])
     async def test_new_and_fork_require_a_declared_checkout(
-        self, native, source_programs, operation, parent_arguments,
+        self, native, source_programs, operation, creation_arguments, expected_tags,
     ):
         await native.author_history()
         comms = wire(native.root)
         await cli(source_programs, native.root, 'register', '--name', 'parent',
-                  '--worktree', str(native.project), '--execution', 'native')
+                  '--tags', 'inherited', '--worktree', str(native.project), '--execution', 'native')
         parent = comms.registry.require('parent')
         comms.threads.attach_session(parent, str(native.session), process=parent.process_identity)
-        command = (operation, '--name', 'probe', *parent_arguments)
+        command = (operation, '--name', 'probe', *creation_arguments)
         plain = native.project.parent / 'plain'
         plain.mkdir()
         for project_arguments in ((), ('--project', ''), ('--project', str(plain))):
@@ -66,7 +68,9 @@ async def test_new_and_fork_require_a_declared_checkout(
                                 '--project', str(native.project),
                                 '--model', 'response-local/fixture', '--thinking-level', 'off')
             assert created['created'] == 'probe'
-            assert comms.registry.require('probe').worktree == Checkout(native.project.resolve())
+            child = comms.registry.require('probe')
+            assert child.worktree == Checkout(native.project.resolve())
+            assert child.tags == expected_tags
         finally:
             for thread in comms.registry.snapshot().threads.values():
                 if thread.process_alive:
diff --git a/tests/test_thread_creation.py b/tests/test_thread_creation.py
new file mode 100644
index 0000000000..e8e82e800d
--- /dev/null
+++ b/tests/test_thread_creation.py
@@ -0,0 +1,130 @@
+"""Creation contracts through the actual CLI decoder, editor binding and registry.
+
+Author does not execute these controls; the integration owner selects them once.
+"""
+from dataclasses import replace
+import json
+
+import pytest
+
+from agent_comms.cli import build_parser
+from agent_comms.channel_targets import CommandersTag
+from agent_comms.cli_command import PendingCommandBinding
+from agent_comms.cli_verbs.fork import ForkCliCommand
+from agent_comms.cli_verbs.new import NewThreadCliCommand
+from agent_comms.nominal.field_codec import MalformedRecord, compile_decoder
+from agent_comms.target_fields import UnsuppliedTargetField
+from agent_comms.launch_runtime import LaunchRuntime
+from agent_comms.repo import Checkout
+from agent_comms.thread_management import GivenTags, InheritedTags
+
+
+def parsed(arguments):
+    namespace = build_parser(arguments).parse_args(arguments)
+    return namespace.command_declaration.from_arguments(namespace)
+
+
+def test_new_cli_requires_an_explicit_tag_argument(source_repository):
+    with pytest.raises(MalformedRecord):
+        parsed(['new', '--name', 'child', '--project', str(source_repository.checkout)])
+
+
+@pytest.mark.parametrize('text,expected', (
+    ('auth,review', frozenset({'auth', 'review'})),
+    ('', frozenset()),
+))
+def test_new_cli_membership_reaches_registration(comms, source_repository, text, expected):
+    command = parsed(['new', '--name', 'child', '--project', str(source_repository.checkout),
+                      '--tags', text])
+    assert command.tags == GivenTags(expected)
+    # New's declaration path reserves its session but does not launch a provider.
+    child = command.specification().declare(comms.threads, '', LaunchRuntime.running())
+    assert child.tags == expected
+    assert comms.registry.require(child.name).tags == expected
+
+
+def test_pending_new_binding_offers_missing_tags_without_constructing_the_command(comms, source_repository):
+    binding, = NewThreadCliCommand.project_bindings(comms, Checkout(source_repository.checkout))
+    assert type(binding) is PendingCommandBinding
+    name = next(item for item in binding.editable_fields if item.name == 'name')
+    assert type(name) is UnsuppliedTargetField
+    assert name.argument(None) == {}
+    with pytest.raises(MalformedRecord):
+        binding.edited(NewThreadCliCommand.editor_arguments({'tags': ''}))
+    tags = next(item for item in binding.editable_fields if item.name == 'tags')
+    assert tags.required
+    assert type(tags) is UnsuppliedTargetField
+    assert 'editor_default' not in tags.json_schema()
+    assert tags.argument(None) == {}
+    assert tags.argument('') == {'tags': ''}
+    with pytest.raises(MalformedRecord):
+        binding.edited(NewThreadCliCommand.editor_arguments({'name': 'child'}))
+
+
+@pytest.mark.parametrize('text,expected', (
+    ('auth', frozenset({'auth'})),
+    ('', frozenset()),
+))
+def test_pending_new_binding_distinguishes_explicit_empty_from_missing(comms, source_repository, text, expected):
+    binding, = NewThreadCliCommand.project_bindings(comms, Checkout(source_repository.checkout))
+    command = binding.edited(NewThreadCliCommand.editor_arguments({'name': 'child', 'tags': text}))
+    assert command.tags == GivenTags(expected)
+    assert command.project == Checkout(source_repository.checkout)
+
+
+def test_pending_catalog_context_can_change_before_tags_are_supplied(comms, source_repository):
+    binding, = NewThreadCliCommand.project_bindings(comms, Checkout(source_repository.checkout))
+    prepared = binding.for_fields({'task': 'child task', 'name': 'child'})
+    assert prepared.context.task == 'child task'
+    assert type(next(item for item in prepared.editable_fields if item.name == 'name')) is UnsuppliedTargetField
+    tags = next(item for item in prepared.editable_fields if item.name == 'tags')
+    assert tags.argument(None) == {}
+    assert tags.argument('') == {'tags': ''}
+
+
+@pytest.mark.parametrize('creation_arguments,forbidden', (
+    (('new', '--tags', 'auth'), {'not_a_field': 'value'}),
+    (('fork', '--parent', 'parent'), {'not_a_field': 'value'}),
+    (('fork', '--parent', 'parent'), {'parent': 'different-parent'}),
+))
+def test_complete_creation_commands_keep_command_edit_authorization(source_repository, creation_arguments, forbidden):
+    command = parsed([*creation_arguments, '--name', 'child',
+                      '--project', str(source_repository.checkout)])
+    with pytest.raises(ValueError):
+        command.edited(forbidden)
+
+
+def test_structured_editor_reports_malformed_json_at_the_data_boundary():
+    decoder = compile_decoder(tuple[int, ...])
+    with pytest.raises(MalformedRecord) as refused:
+        decoder.decode_editor('[1,')
+    assert type(refused.value.__cause__) is json.JSONDecodeError
+    assert decoder.decode(decoder.decode_editor('[1, 2]')) == (1, 2)
+
+
+def test_fork_cli_defaults_to_actual_parent_membership(comms, source_repository):
+    parent_command = parsed(['new', '--name', 'parent', '--project', str(source_repository.checkout),
+                             '--tags', 'original'])
+    parent = parent_command.specification().declare(comms.threads, '', LaunchRuntime.running())
+    command = parsed(['fork', '--name', 'child', '--parent', parent.name,
+                      '--project', str(source_repository.checkout)])
+    assert command.tags == InheritedTags()
+    binding = ForkCliCommand.thread_binding(parent)
+    assert type(binding) is PendingCommandBinding
+    assert binding.context.parent == parent.name
+    assert type(next(item for item in binding.editable_fields if item.name == 'name')) is UnsuppliedTargetField
+    with pytest.raises(MalformedRecord):
+        binding.edited({})
+    with pytest.raises(ValueError, match='Target-bound'):
+        binding.edited({'name': 'child', 'parent': 'other'})
+    editor = binding.edited(ForkCliCommand.editor_arguments({'name': 'child'}))
+    assert editor.tags == InheritedTags()
+    # A real later parent record, not the stale tags at editor opening, supplies membership.
+    # The commander role is declared, never inherited; an explicit choice keeps it.
+    commander = CommandersTag.declared_name
+    current = comms.registry.declare(replace(parent, tags=frozenset({'current', commander})))
+    child = command.specification().child(current, current.runtime, comms.threads)
+    assert child.tags == current.tags - {commander} == frozenset({'current'})
+    explicit = replace(command, tags=GivenTags(current.tags))
+    assert commander in explicit.specification().child(current, current.runtime, comms.threads).tags
+    assert editor.tags.for_child(current.tags) == current.tags