class-node.php

<?php
/**
 * The contract every node in the graph honors, and the machinery each one
 * inherits rather than rebuilds: naming, owned siblings, event channels and
 * rate-limited diagnostics.
 *
 * @package Newspack_Nodes
 */

namespace Newspack_Nodes;

\defined( 'ABSPATH' ) || exit;

/**
 * A processing unit in the message graph.
 *
 * `fill()` is the whole contract — there is no `write()`, `read()` or
 * `process()` (ADR-1) — which is what lets any node compose with any other. A
 * node connects two ways (ADR-7): `sink` is the physical next hop `fill()`
 * forwards to, and `target` is the logical path stamped into TO when TO is
 * empty, which Router then peels one segment at a time. There is no `edge`.
 *
 * The base `fill()` stamps TO, counts and forwards, so a subclass that
 * overrides it calls `parent::fill()` to keep that. Everything else here is
 * shared: registry naming and its rename cascade, the owned-sibling quartet,
 * the `register()` / `notify()` / `set_state()` event channels, and the
 * throttled stderr diagnostics.
 */
class Node {

	/** Stand-in for a masked credential in diagnostics. */
	public const REDACTED = '<redacted>';

	/**
	 * Byte ceiling on a message's FROM path. `stamp_message()` refuses to stamp
	 * past it and Router drops past it, which is what keeps a routing cycle from
	 * growing the breadcrumb trail without bound.
	 */
	public const MAX_FROM_SIZE = 1024;

	/** The sibling kind of a node's `{name}:config` interpreter. */
	public const CONFIG_KIND = 'config';

	/** Message types whose payload is included in the drop_message() audit line. */
	private const PAYLOAD_TYPES = Message::TM_INFO | Message::TM_REQUEST | Message::TM_ERROR | Message::TM_COMMAND;

	/** @var list<string> Constructor argument tokens (Tachikoma's raw `arguments`). */
	protected array $arguments = [];

	/**
	 * Bytes pulled off a stream, for the `ls -l` and `dump_metadata` columns.
	 * Only a node that reads one advances it — Partition, Consumer, File_Tail and
	 * SSE_In — so a pure logic node reports zero.
	 */
	protected int $bytes_read = 0;

	/** Bytes pushed to a stream; the `File_Writer` trait is what advances it. */
	protected int $bytes_written = 0;

	/** Messages this node has filled — the `ls -c` and `dump_metadata` counter. */
	protected int $counter = 0;

	/**
	 * Per-node trace dial, written by the REPL's `trace` command: 0 is quiet, and
	 * anything higher makes `set_state()` also emit a flat `DEBUG: <event>
	 * <payload>` stderr line, which the console's timeline reads.
	 */
	protected int $debug_state = 0;

	/** Sibling CommandInterpreter (`:config`) for nodes with runtime config commands; else null. */
	protected ?Command_Interpreter_Node $interpreter = null;

	/** Largest single message this node has moved, in bytes; a stats column only. */
	protected int $largest_msg_sent = 0;

	/**
	 * The registry key `Core::node()` resolves and Router matches each TO segment
	 * against. Empty until `name()` registers the node.
	 */
	protected string $name = '';

	/** Non-null marks this node as the patron's plumbing; see `owned()`. */
	protected ?Node $patron = null;

	/** The node that published this one as an owned sibling, or null. */
	private ?Node $publisher = null;

	/** False once a builder opts this node out; see `without_config()`. */
	private bool $wants_config = true;

	/**
	 * Declared events, each holding its listeners keyed by identity. A closure is
	 * stored as given; a null value means the identity is a Node name and dispatch
	 * goes out as TM_INFO. `seed_registrations()` fills the event keys from
	 * `node_schema()`, and `register()` refuses anything not among them, so this
	 * map is the node's published event contract.
	 *
	 * @var array<string,array<string,callable|string|null>>
	 */
	protected array $registrations = [];

	/**
	 * Latest payload per event, so a listener that registers afterwards gets
	 * current state at once instead of waiting for the next `notify()`.
	 *
	 * @var array<string,string>
	 */
	protected array $set_state = [];

	/**
	 * The nodes this one built for itself, keyed by the KIND each was built for
	 * — the collision pre-check, the rename, the sink and the teardown all walk
	 * this, and `siblings()` hands a subclass a read of it. Written only by
	 * `publish_sibling()` / `retract_sibling()`, so publishing IS declaring and
	 * there is no declaration to forget.
	 *
	 * @var array<string,Node>
	 */
	private array $siblings = [];

	/** The physical next hop `fill()` forwards to; `target` is the logical half. */
	protected ?Node $sink = null;

	/**
	 * Closures this node listens with at OTHER nodes, by emitter name and then
	 * event. The emitter keys each under this node's name, so `name()` moves
	 * them and `remove_node()` drops them.
	 *
	 * @var array<array-key,array<string,\Closure>>
	 */
	private array $subscriptions = [];

	/**
	 * The logical path stamped into an empty TO. A string names one destination;
	 * the array form is Tee's fan-out, so an array answer means this node fans out.
	 *
	 * @var string|array<int,string>
	 */
	protected $target = '';

	/**
	 * Seed the declared-event allow-list, and stand as the chain anchor.
	 *
	 * A node acquires schema-reflection behavior — positional argument parsing, the
	 * `{name}:config` interpreter auto-wire — only by `use`-ing the
	 * Schema_Reflection trait and calling its helpers; Node itself carries none.
	 * Declaring the constructor here lets any subclass call `parent::__construct()`
	 * whatever the intermediate classes do, and node-specific setup stays in the
	 * subclass constructor.
	 */
	public function __construct() {
		$this->seed_registrations();
	}

	/**
	 * Get/set the node's raw argument tokens — Tachikoma's plain getter/setter,
	 * which does NOT parse them. A node wanting positional config uses the
	 * Schema_Reflection trait, whose arguments() runs the tokens through
	 * parse_schema_args(); only a node deriving state from them overrides it.
	 *
	 * @param list<string>|null $args New argument tokens (null = pure getter).
	 * @return list<string> Last-set argument tokens.
	 */
	public function arguments( ?array $args = null ): array {
		if ( null !== $args ) {
			$this->arguments = $args;
		}
		return $this->arguments;
	}

	/**
	 * Default: stamp TO from target if empty, then forward to sink.
	 *
	 * @param array<int,mixed> $message The 7-field positional message array.
	 */
	public function fill( array $message ): void {
		$sink = $this->require_sink();
		if ( '' === $message[ Message::TO ] && \is_string( $this->target ) && '' !== $this->target ) {
			$message[ Message::TO ] = $this->target;
		}
		++$this->counter;
		$sink->fill( $message );
	}

	/**
	 * The wired sink, or a loud refusal — a node with no sink cannot emit.
	 *
	 * @throws \RuntimeException When the node has no sink.
	 */
	protected function require_sink(): Node {
		return $this->sink ?? throw new \RuntimeException( 'fill requires a wired sink' );
	}

	/**
	 * Read the node's name, or register it under a new one.
	 *
	 * A set pre-checks every slot the rename will touch, so a collision throws
	 * before anything moves; the old registration is then released, the new one
	 * taken, and every owned sibling renamed to match. Re-setting the name the
	 * node already answers to is a no-op rather than a collision with itself.
	 * An empty name is refused because `remove_node()` is how a node leaves the
	 * registry, and one that stayed registered without a name is unreachable.
	 *
	 * @param string|null $name New name; omit the argument entirely to read.
	 * @return string The name the node answers to.
	 * @throws \RuntimeException When the name is empty, or a slot is taken.
	 */
	public function name( ?string $name = null ): string {
		if ( \func_num_args() > 0 ) {
			if ( ! Core::has_value( $name ) ) {
				throw new \RuntimeException( 'name() requires a non-empty name; use remove_node() to unregister' );
			}
			if ( $name === $this->name ) {
				return $this->name;
			}
			$this->check_name_availability( $name );
			$previous = $this->name;
			if ( '' !== $previous ) {
				Core::unregister_node( $previous );
			}
			$this->name = $name;
			Core::register_node( $name, $this );
			$this->set_sibling_names();
			$this->move_subscriptions( $previous, $name );
		}
		return $this->name;
	}

	/**
	 * Pre-check each owned sibling's slot and then the node's own, so a
	 * collision throws before anything is registered. The recursion has to
	 * reach exactly as far as the rename cascade does, since
	 * `set_sibling_names()` calls `$sibling->name()`: stop at the direct slots
	 * and a deeper squat throws from the middle of a half-applied rename.
	 * Siblings rank first because a squatted sibling slot is the harder
	 * collision to diagnose from outside — the patron's own name is right there
	 * in the topology — and the constructor-published `:config` interpreter
	 * ranks LAST, after the node's own name too, despite coming first by
	 * insertion order, derived scaffolding being the least useful collision to
	 * report. An owned sibling keeps a `:config` of its own, so without that a
	 * squatted `{name}:source` would report as its `:config`.
	 *
	 * It checks the siblings that EXIST, so it covers a RENAME fully and a lazy
	 * patron's FIRST naming not at all: `make_node` runs `name()` before
	 * `arguments()`, so `Topic_Node` cannot know its partition count that early
	 * and has nothing to check. That collision surfaces a step later, from the
	 * sibling's own `name()`; `make_node` unwinds either way, so only the
	 * timing differs.
	 *
	 * Pre-checking rather than unwinding is sound only because the registry is
	 * a per-process array mutated synchronously (`Core::register_node()`), so
	 * a name can never be taken between the check and the cascade. Give it an
	 * out-of-process source and this becomes TOCTOU and the unwind returns.
	 *
	 * @param string $name Name the node is about to take.
	 * @throws \RuntimeException When the node's or a sibling's slot is taken,
	 *                           or either name holds the Router's `/`.
	 */
	protected function check_name_availability( string $name ): void {
		if ( \str_contains( $name, '/' ) ) {
			throw new \RuntimeException( \esc_html( "node name {$name} holds a /, which the Router reads as a path separator" ) );
		}
		// `workers inspect` trusts a `:config` name to address an interpreter.
		$config = self::sibling_name_of( '', self::CONFIG_KIND );
		if ( \str_ends_with( $name, $config ) && ! $this instanceof Command_Interpreter_Node ) {
			throw new \RuntimeException( \esc_html( "node name {$name} ends {$config}, which only a config interpreter takes" ) );
		}
		$deferred = $this->siblings[ self::CONFIG_KIND ] ?? null;
		foreach ( $this->siblings as $kind => $sibling ) {
			if ( $sibling !== $deferred ) {
				$sibling->check_name_availability( self::sibling_name_of( $name, $this->sibling_suffix( $kind ) ) );
			}
		}
		if ( Core::node( $name ) !== null ) {
			throw new \RuntimeException( \esc_html( "node name collision: {$name} already registered" ) );
		}
		$deferred?->check_name_availability( self::sibling_name_of( $name, $this->sibling_suffix( self::CONFIG_KIND ) ) );
	}

	/**
	 * Publish a FINISHED sibling into the slot it was built for, then name it
	 * `{name}:{suffix}`. Publishing is what enrols the sibling in the four
	 * cascades — the collision pre-check, the rename, the sink and teardown —
	 * so a builder that publishes cannot leave one unnamed, unreachable from
	 * the registry, `ls` and `command_node`, and still writing to disk.
	 *
	 * A refused name empties the slot again, so a caller's idempotency guard
	 * rebuilds rather than serving a half-published sibling.
	 *
	 * An OCCUPIED slot is refused rather than overwritten: overwriting strands
	 * the displaced node, still registered and still holding whatever handle it
	 * opened, but not reached by any cascade. A publisher that means to REPLACE
	 * empties the slot with `retract_sibling()` first.
	 *
	 * @param string $kind    What the builder built; the slot key.
	 * @param Node   $sibling The finished sibling.
	 * @throws \RuntimeException When the slot is occupied or the name is refused.
	 */
	protected function publish_sibling( string $kind, Node $sibling ): void {
		if ( isset( $this->siblings[ $kind ] ) ) {
			throw new \RuntimeException( \esc_html( "sibling slot occupied: {$kind}; retract_sibling() to replace it" ) );
		}
		$this->siblings[ $kind ] = $sibling;
		$sibling->publisher      = $this;
		try {
			$this->set_sibling_names();
		} catch ( \Throwable $e ) {
			unset( $this->siblings[ $kind ] );
			$sibling->publisher = null;
			throw $e;
		}
	}

	/**
	 * Name every owned sibling `{name}:{suffix}`, resolving the suffix from the
	 * slot's stable KIND — the ONE place a sibling's name is spelled, reached
	 * from `name()` on a rename, from `publish_sibling()` on a new slot and
	 * from a schema-arg replay, so a re-keyed suffix cannot leave a sibling
	 * answering to a superseded name. One live exception: `Raw_Logs_CI_Node`
	 * spells `{name}:status` inline for a probe it builds, reads and tears down
	 * inside a single command call, which a rename can never reach. A no-op while
	 * the patron is unnamed, which is what lets a builder publish from a
	 * constructor.
	 */
	protected function set_sibling_names(): void {
		if ( '' === $this->name ) {
			return;
		}
		foreach ( $this->siblings as $kind => $sibling ) {
			$sibling->name( self::sibling_name_of( $this->name, $this->sibling_suffix( $kind ) ) );
		}
	}

	/**
	 * Prepend $name to the message's FROM path, so a reply can walk the trail
	 * back. Refused two ways, and the caller must drop the message on either: an
	 * empty $name would compose a `/from` path Router cannot resolve, and a path
	 * past MAX_FROM_SIZE means a cycle is growing the trail.
	 *
	 * @param array<int,mixed> $message The 7-field message array, stamped in place.
	 * @param string           $name    The name to prepend.
	 * @return bool False when the stamp was refused.
	 */
	public function stamp_message( array &$message, string $name ): bool {
		if ( '' === $name ) {
			$this->print_less_often( 'ERROR: ' . static::class . ' stamp_message() called with empty name' );
			return false;
		}
		$new = self::stamped_path( $message, $name );
		if ( \strlen( $new ) > self::MAX_FROM_SIZE ) {
			$this->print_less_often( 'ERROR: path exceeded ' . self::MAX_FROM_SIZE . ' bytes; dropping from: ', $new );
			return false;
		}
		$message[ Message::FROM ] = $new;
		return true;
	}

	/**
	 * Whether `stamp_message()` would take this message — asked WITHOUT the
	 * warning, for a boundary that reports the refusal some other way.
	 *
	 * `HTTP_In_Node` is the one: a drop's stderr line routes into `_output` once
	 * a request graph exists, which opens the response body and spends the
	 * status line, so it asks first and answers the client with a frame.
	 *
	 * @param array<int,mixed> $message The 7-field message array.
	 * @param string           $name    The name that would be prepended.
	 * @return bool False when the stamp would be refused.
	 */
	public static function can_stamp( array $message, string $name ): bool {
		return '' !== $name && \strlen( self::stamped_path( $message, $name ) ) <= self::MAX_FROM_SIZE;
	}

	/**
	 * The FROM path `$name` would compose, so the check and the write agree.
	 *
	 * @param array<int,mixed> $message The 7-field message array.
	 * @param string           $name    The name to prepend.
	 */
	private static function stamped_path( array $message, string $name ): string {
		$from = Core::as_string( $message[ Message::FROM ] );
		return '' === $from ? $name : ( $name . '/' . $from );
	}

	/**
	 * Multi-modal listener: store either a closure (with callable) or a Node name string.
	 *
	 * The entry lives in the emitter's table, so the listener's rename is
	 * never followed here. A node listening with a closure under its OWN name
	 * goes through `subscribe()`, which does; `Timer_Node` moves its
	 * name-dispatched TIMER hitchhike itself, in its own `name()`.
	 *
	 * @param string        $event    Must be pre-declared in registrations.
	 * @param string        $listener Identity (closure ID or Node name).
	 * @param callable|null $cb       Closure. If null, $listener is a Node name.
	 * @throws \RuntimeException When the event was never declared.
	 */
	public function register( string $event, string $listener, ?callable $cb = null ): void {
		if ( ! isset( $this->registrations[ $event ] ) ) {
			throw new \RuntimeException( \esc_html( "no such event: $event" ) );
		}
		$this->registrations[ $event ][ $listener ] = $cb;

		if ( \array_key_exists( $event, $this->set_state ) ) {
			$this->_notify_registered( $event, $listener, $this->set_state[ $event ] );
		}
	}

	/**
	 * Cache the payload under $event and notify every listener; the cache is what
	 * lets a late registrant receive current state at register-time. With
	 * `debug_state` on and a `_router` mounted, also emit a flat Tachikoma-style
	 * `DEBUG: <event> <payload>` line.
	 *
	 * @param string $event   Event name. Caching under one declares nothing, so an
	 *                        undeclared name reaches `dump_node` and the trace line
	 *                        while accepting no subscribers.
	 * @param string $payload Current state.
	 */
	protected function set_state( string $event, string $payload = '' ): void {
		$this->set_state[ $event ] = $payload;
		if ( $this->debug_state > 0 ) {
			$router = Core::node( Node_Names::ROUTER );
			if ( null !== $router ) {
				$detail = Core::as_string( $payload );
				$this->stderr( 'DEBUG: ' . $event . ( '' !== $detail ? ' ' . $detail : '' ) );
			}
		}
		$this->notify( $event, $payload );
	}

	/**
	 * Fire the event at every currently-registered listener. A closure returning
	 * exactly false is dropped, which is the only self-heal a closure gets. The
	 * walk is over a snapshot, so a listener an earlier one unregistered during
	 * this dispatch is skipped rather than reported as forgotten.
	 *
	 * @param string $event   Event name; an undeclared one is a silent no-op.
	 * @param mixed  $payload Handed to a closure, carried as a Node name's TM_INFO VALUE.
	 */
	public function notify( string $event, mixed $payload = '' ): void {
		if ( ! isset( $this->registrations[ $event ] ) ) {
			return;
		}
		foreach ( $this->registrations[ $event ] as $listener => $cb ) {
			$keep = $this->_notify_registered( $event, $listener, $payload );
			if ( false === $keep ) {
				unset( $this->registrations[ $event ][ $listener ] );
			}
		}
	}

	/**
	 * Dispatch one listener: a closure takes the payload directly, and a Node name
	 * takes a TM_INFO message with KEY=event.
	 *
	 * @param string $event    Event name.
	 * @param string $listener Closure ID or Node name.
	 * @param mixed  $payload  Event payload.
	 * @return mixed The closure's own return; for a Node name, false when that node
	 *               is gone and true otherwise. Exactly false unregisters the listener.
	 *               Null for a listener no longer registered, which is skipped.
	 */
	private function _notify_registered( string $event, string $listener, mixed $payload ): mixed {
		// Unregistered mid-dispatch: notify() walks a snapshot of the table.
		if ( ! \array_key_exists( $listener, $this->registrations[ $event ] ?? [] ) ) {
			return null;
		}
		$cb = $this->registrations[ $event ][ $listener ] ?? null;
		if ( null !== $cb && \is_callable( $cb ) ) {
			return $cb( $payload );
		}
		$target = Core::node( $listener );
		if ( null === $target ) {
			$this->print_less_often( "WARNING: $listener forgot to unregister from $event on " . $this->name );
			return false;
		}
		$message                   = Message::new_message();
		$message[ Message::TYPE ]  = Message::TM_INFO;
		$message[ Message::FROM ]  = $this->name;
		$message[ Message::KEY ]   = $event;
		$message[ Message::VALUE ] = $payload ?? '';
		$target->fill( $message );
		return true;
	}

	/**
	 * Drop a message, leaving one rate-limited audit line that names its type
	 * flags, FROM, TO and — for the four control types — its redacted payload.
	 *
	 * The throttle keys on $error alone. $tail prints last and stays out of
	 * the key, so a caller can name a detail that varies per message without
	 * multiplying the lines.
	 *
	 * @param array<int,mixed> $message The 7-field positional message array.
	 * @param string           $error   Reason; `NOT_AVAILABLE` prints unprefixed.
	 * @param string           $tail    Detail printed after the payload; '' for none.
	 */
	public function drop_message( array $message, string $error, string $tail = '' ): void {
		$type_raw = $message[ Message::TYPE ];
		$type     = Core::num_int( $type_raw );
		$labels   = Message::type_labels( $type );
		$type_str = empty( $labels ) ? 'TYPE_UNKNOWN' : \implode( '|', $labels );
		$prefix   = 'NOT_AVAILABLE' === $error ? $error : "WARNING: $error";
		$parts    = [ "$prefix -", $type_str ];
		$from = Core::as_string( $message[ Message::FROM ] );
		if ( '' !== $from ) {
			$parts[] = 'from: ' . $from;
		}
		$to = Core::as_string( $message[ Message::TO ] );
		if ( '' !== $to ) {
			$parts[] = 'to: ' . $to;
		}
		$value = $message[ Message::VALUE ];
		if ( ( $type & self::PAYLOAD_TYPES ) && '' !== $value ) {
			$redacted = self::redact_secrets( $value );
			$shown    = ( $type & Message::TM_COMMAND )
				? self::command_summary( $redacted, Message::is_request_command( $type ) )
				: $redacted;
			$value_str = \is_array( $shown )
				? (string) \wp_json_encode( $shown, \JSON_UNESCAPED_SLASHES | \JSON_INVALID_UTF8_SUBSTITUTE )
				: Core::as_string( $shown );
			$parts[] = 'payload: ' . $value_str;
		}
		if ( '' !== $tail ) {
			$parts[] = $tail;
		}
		$head = \array_shift( $parts );
		$this->print_less_often( $head, ' ' . \implode( ' ', $parts ) );
	}

	/**
	 * A command as the audit line says it: the verb and its arguments, which is
	 * `$command->name . q( ) . $command->arguments` in Tachikoma's
	 * `drop_message`.
	 *
	 * The rest of a TM_COMMAND VALUE is the signing envelope — nonce,
	 * signature, session handle — and none of it says what was lost. Encoding
	 * the whole VALUE put all three on stderr, and on the REST command path
	 * into the response body, where the `SECRET_NAME_PATTERNS` entries do not
	 * match to mask them. The command's own `payload` goes with them: Tachikoma
	 * prints neither, and a credential reaches a command through it as readily as
	 * through an argument.
	 *
	 * The caller redacts the WHOLE VALUE first, so a secret keyed by name at
	 * any depth of `arguments` is masked as surely as a `--password=…` token.
	 * This is the one place a VALUE's arguments become string tokens: a keyed
	 * map yields its values, and a non-scalar yields ''. `commandSummary()` in
	 * `src/runtime/node.js` prints the same line.
	 *
	 * A request's array VALUE that does not name a verb is summarised all the
	 * same, its verb '', because a request is what carries the envelope; a
	 * response naming none prints whole.
	 *
	 * @param mixed $value   The redacted message VALUE.
	 * @param bool  $request Whether the message is a request, not a reply.
	 * @return mixed `name arguments`, each token quoted by `serialize_args()`,
	 *               for a request's array VALUE or a named one; else it unchanged.
	 */
	private static function command_summary( mixed $value, bool $request ): mixed {
		if ( ! \is_array( $value ) || ( ! $request && ! \array_key_exists( 'name', $value ) ) ) {
			return $value;
		}
		$arguments = \is_array( $value['arguments'] ?? null ) ? $value['arguments'] : [];
		$tokens    = \array_map( static fn ( $token ): string => Core::as_string( $token ), \array_values( $arguments ) );
		return self::serialize_args( [ Core::as_string( $value['name'] ?? '' ), ...$tokens ] );
	}

	/**
	 * Emit a stderr line tagged with this node's midfix, via Core's stderr
	 * pipeline (which adds the process-identity midfix and timestamp prefix once).
	 * Empty text is a no-op (Tachikoma Node::stderr).
	 */
	public function stderr( string $text ): void {
		if ( '' === $text ) {
			return;
		}
		Core::stderr( $this->log_midfix( $text ) );
	}

	/**
	 * Emit text on first sight and suppress it until `Core::prune_logs()`
	 * re-windows the key, so one high-volume error path cannot flood stderr. The
	 * key is the midfixed $text alone — per-node, sharing
	 * `Core::$recent_log_timers` — so a category whose detail varies still
	 * collapses to one line rather than one line per value.
	 *
	 * @param string $text     Throttle key, and the head of the line.
	 * @param string ...$extra Payload, appended only on the occurrence that prints.
	 */
	public function print_less_often( string $text, string ...$extra ): void {
		$key = $this->log_midfix( $text );
		$timestamp = Core::$recent_log_timers[ $key ] ?? null;
		if ( null === $timestamp ) {
			Core::stderr( $this->log_midfix( $text . \implode( '', $extra ) ) );
			Core::$recent_log_timers[ $key ] = Core::$now;
		}
	}

	/**
	 * Per-node mid-line tag: "<name>: " prepended to every line. Empty when
	 * the node is unnamed, or when the process identity ($0 / Core::argv0())
	 * already starts with the node name (so the tag would be redundant). With
	 * text, chomps a trailing newline, prepends the tag to every line,
	 * and appends one trailing newline.
	 *
	 * @param string|null $text Text to tag; null yields the bare midfix.
	 * @return string The tagged text, or the midfix alone.
	 */
	public function log_midfix( ?string $text = null ): string {
		$midfix = '';
		if ( '' !== $this->name
			&& 1 !== \preg_match( '/^' . \preg_quote( $this->name, '/' ) . '\b/', Core::argv0() ) ) {
			$midfix = $this->name . ': ';
		}
		return Core::apply_midfix( $midfix, $text );
	}

	/**
	 * Seed the runtime registration allow-list from node_schema()['registrations']
	 * — the single source of valid events. A registration-capable node calls this
	 * in its constructor instead of hand-assigning $this->registrations.
	 */
	protected function seed_registrations(): void {
		$events = static::node_schema()['registrations'] ?? [];
		if ( ! \is_array( $events ) ) {
			return;
		}
		foreach ( $events as $event ) {
			if ( \is_string( $event ) ) {
				$this->registrations[ $event ] = [];
			}
		}
	}

	/**
	 * Round-trippable graph snippet: a `make_node` line, then `set_sink` and
	 * `connect_node` where they differ from the defaults. `set_sink` is suppressed
	 * for `_command_interpreter`, which every node sinks into anyway.
	 *
	 * @return string Newline-terminated TSL lines.
	 */
	public function dump_config(): string {
		$short = Command_Interpreter_Node::shell_name_for( $this );
		$out   = self::command_line( 'make_node', $short, $this->name, ...$this->arguments );
		if ( null !== $this->sink ) {
			$sink_name = $this->sink->name();
			if ( '' !== $sink_name && Node_Names::COMMAND_INTERPRETER !== $sink_name ) {
				$out .= self::command_line( 'set_sink', $this->name, $sink_name );
			}
		}
		if ( \is_array( $this->target ) ) {
			foreach ( $this->target as $owner ) {
				$out .= self::command_line( 'connect_node', $this->name, $owner );
			}
		} elseif ( '' !== $this->target ) {
			$out .= self::command_line( 'connect_node', $this->name, $this->target );
		}
		return $out;
	}

	/**
	 * A `command_node <name>:<config-slot> …` line addressed to this node's own
	 * sibling interpreter.
	 *
	 * @param string ...$tokens The verb and its arguments.
	 * @return string One newline-terminated TSL line.
	 */
	protected function config_line( string ...$tokens ): string {
		return self::command_line( 'command_node', $this->sibling_name( self::CONFIG_KIND ), ...$tokens );
	}

	/**
	 * One replayable TSL line from its tokens. A command line IS an argv — the
	 * verb, the type, the name and the arguments are all just tokens, so every
	 * one of them goes through the same quoting. `make_node Echo echo foo bar`
	 * reads as if `echo` were the name and `foo bar` the arguments; the command's
	 * actual arguments are all four.
	 *
	 * @param string ...$tokens The whole argv — verb, type, name and arguments alike.
	 * @return string One newline-terminated line.
	 */
	public static function command_line( string ...$tokens ): string {
		return self::serialize_args( \array_values( $tokens ) ) . "\n";
	}

	/**
	 * Serialize argument tokens back to a single line. The one place tokens are
	 * re-joined — every other layer carries them as a list.
	 *
	 * @param list<string> $tokens Tokens to quote and join.
	 * @return string The tokens re-joined with single spaces.
	 */
	public static function serialize_args( array $tokens ): string {
		return \implode( ' ', \array_map( [ self::class, 'serialize_arg' ], $tokens ) );
	}

	/**
	 * Snapshot of this node's state for the REPL `dump_node` command. Credentials
	 * are masked by the same one rule the drop audit uses — redact_secrets(),
	 * which reaches a secret nested inside an ordinary property (an
	 * `--password=…` token in $arguments) as well as a secret-named one.
	 *
	 * @return array<string,mixed>
	 */
	public function dump_node(): array {
		$ref      = new \ReflectionObject( $this );
		$snapshot = [];
		foreach ( $ref->getProperties() as $prop ) {
			$key   = $prop->getName();
			if ( $prop->isInitialized( $this ) ) {
				$value = $prop->getValue( $this );
			} else {
				$value = 'null';
			}
			if ( 'sink' === $key && $value instanceof Node ) {
				$value = $value->name();
			}
			if ( Core::is_secret_property( $key )
				&& ( ( \is_string( $value ) && '' !== $value ) || ( \is_array( $value ) && [] !== $value ) ) ) {
				$value = self::REDACTED;
			} else {
				$value = self::redact_secrets( $value );
			}
			if ( \is_object( $value ) ) {
				$value = '(' . \get_class( $value ) . ')';
			}
			if ( \is_resource( $value ) ) {
				$value = '(resource:' . \get_resource_type( $value ) . ')';
			}
			$snapshot[ $key ] = $value;
		}
		$snapshot['class'] = $ref->getShortName();
		return $snapshot;
	}

	/**
	 * Mask credentials in a dropped message's VALUE, by the same rule
	 * `dump_node()` uses — `Core::is_secret_property()`. Two shapes carry them:
	 * a secret-named array key, and a `--password=…` argument token, which
	 * is how the Vault admin UI sends them. The key survives; only the value
	 * goes, because the drop line is a diagnostic.
	 *
	 * @param mixed $value Message VALUE, any depth.
	 * @return mixed The same shape with secrets masked.
	 */
	private static function redact_secrets( $value ) {
		if ( \is_array( $value ) ) {
			$out = [];
			foreach ( $value as $key => $item ) {
				$out[ $key ] = \is_string( $key ) && Core::is_secret_property( $key )
					? self::REDACTED
					: self::redact_secrets( $item );
			}
			return $out;
		}
		if ( \is_string( $value ) && \str_starts_with( $value, '--' ) ) {
			$eq = \strpos( $value, '=' );
			if ( false !== $eq && Core::is_secret_property( \substr( $value, 2, $eq - 2 ) ) ) {
				return \substr( $value, 0, $eq + 1 ) . self::REDACTED;
			}
		}
		return $value;
	}

	/**
	 * Whether a patron owns this node, which keeps it off the console canvas:
	 * the console opens it from its owner's `sidecars()` instead.
	 */
	public function owned(): bool {
		return null !== $this->patron();
	}

	/**
	 * The owned siblings the console opens from this node, sorted by name:
	 * every published sibling with a patron but the `:config` interpreter,
	 * whose commands the node's own Inspector already sends, and any whose schema
	 * is `hidden`. A published sibling with no patron, such as a Vault_Group's
	 * child, draws on the canvas instead.
	 *
	 * @return list<array{name: string, class: string}>
	 */
	public function sidecars(): array {
		$out = [];
		foreach ( $this->siblings as $kind => $sibling ) {
			if ( self::CONFIG_KIND !== $kind && $sibling->owned() && ! self::hidden( $sibling::node_schema() ) ) {
				$out[] = [
					'name'  => $sibling->name(),
					'class' => Command_Interpreter_Node::shell_name_for( $sibling ),
				];
			}
		}
		\usort( $out, static fn ( array $a, array $b ): int => \strcmp( $a['name'], $b['name'] ) );
		return $out;
	}

	/**
	 * Patron getter/setter. A patron marks this node owned — off the canvas,
	 * listed in its owner's `sidecars()` — and changes nothing else: the node
	 * keeps its `{name}:config` interpreter, so its commands reach it whoever
	 * owns it, and the patron may be set before or after `name()`.
	 *
	 * @param Node|null $node The patron to adopt (null = pure getter).
	 * @return Node|null The patron, or null when the node stands on its own.
	 */
	public function patron( ?Node $node = null ): ?Node {
		if ( null !== $node ) {
			$this->patron = $node;
		}
		return $this->patron;
	}

	/**
	 * Build this node with no `{name}:config` interpreter, so none of its
	 * commands is reachable. A builder calls it before `name()`, for a
	 * node that is not configurable by any caller: a reader's view of a log it
	 * does not write, or a node a request builds and drops. Refused once named,
	 * because the interpreter is registered by then and an address may already
	 * hold it.
	 *
	 * @return self
	 * @throws \RuntimeException When the node is already named.
	 */
	public function without_config(): self {
		if ( '' !== $this->name ) {
			throw new \RuntimeException(
				\esc_html( static::class . '::without_config() must be called before name(): ' . $this->sibling_name( self::CONFIG_KIND ) . ' is registered already' )
			);
		}
		$this->retract_sibling( self::CONFIG_KIND );
		$this->interpreter  = null;
		$this->wants_config = false;
		return $this;
	}

	/**
	 * The name the sibling in $kind's slot answers to. Anything ADDRESSING a
	 * sibling reads it here rather than spelling the suffix again, so a
	 * re-keyed slot cannot leave a message routed at a name that is not
	 * registered.
	 *
	 * @param string $kind What the builder built, resolved through `sibling_suffix()`.
	 * @return string The sibling's registered name.
	 */
	protected function sibling_name( string $kind ): string {
		return self::sibling_name_of( $this->name, $this->sibling_suffix( $kind ) );
	}

	/**
	 * `{patron}:{suffix}` — the ONE spelling of a sibling's name. The namer,
	 * the collision pre-check, `sibling_name()` and `Topology_Analyzer`'s
	 * flatten of a `Vault_Group` all compose through it.
	 *
	 * @api Topology_Analyzer names a flattened group child through this.
	 * @param string $patron The publishing node's name.
	 * @param string $suffix What `sibling_suffix()` answers for the slot.
	 */
	public static function sibling_name_of( string $patron, string $suffix ): string {
		return "{$patron}:{$suffix}";
	}

	/**
	 * The NAME suffix a builder's $kind takes, `{patron}:{suffix}`. A class
	 * re-spelling an inherited sibling's name overrides this one method — the
	 * rename, the collision pre-check and `sibling_name()` all read it here.
	 * The slot KEY stays the kind and never moves, or a re-spelling would
	 * strand the incumbent in a slot the next retract computes past. The suffix
	 * is interpolated, never parsed, so a COMPOUND one is fine.
	 *
	 * @param string $kind What a builder built — `source`, `offsetlog`, `deadletter`.
	 * @return string The suffix that follows the patron's name.
	 */
	protected function sibling_suffix( string $kind ): string {
		return $kind;
	}

	/**
	 * Every destination the console draws an edge for: the routing target plus
	 * the declared extras, primary first, de-duplicated and without empties.
	 *
	 * @return list<string>
	 */
	public function display_targets(): array {
		$extras = \array_filter( $this->extra_targets(), static fn ( string $extra ): bool => '' !== $extra );
		return \array_values( \array_unique( [ ...self::target_list( $this->target() ), ...$extras ] ) );
	}

	/**
	 * The destinations this node writes to WITHOUT routing through `target` — a
	 * sibling's own target, a partition written straight at flush. The console
	 * draws one edge per entry, so an omitted destination renders disconnected
	 * on the canvas while it fills. Declare the fields; empties and duplicates
	 * are `display_targets()`'s problem, not the declaration's. By default the
	 * ones the class declares for its own arguments.
	 *
	 * @return list<string>
	 */
	protected function extra_targets(): array {
		return static::declared_targets( $this->arguments );
	}

	/**
	 * The destinations a class's `make_node` arguments name for it to write
	 * without routing through `target`: what a running node declares through
	 * `extra_targets()`, and what `Topology_Analyzer` reads off a statement for
	 * every class and draws as `pair` edges, so the two cannot drift. None by
	 * default.
	 *
	 * @param list<string> $args The `make_node` argument tokens, the name excluded.
	 * @return list<string>
	 */
	public static function declared_targets( array $args ): array {
		return [];
	}

	/**
	 * Tear the node down: drop its subscriptions, then cascade a full
	 * `remove_node()` to each owned sibling first so a same-name respawn cannot
	 * collide with a leftover slot. Every sibling is torn down whatever another
	 * threw — a Partition's teardown flushes — and this node's own teardown
	 * still completes; the failures escape after it.
	 *
	 * @throws \Throwable What the siblings' teardowns threw, combined by `Worker_Should_Stop::raise()`.
	 */
	public function remove_node(): void {
		$this->move_subscriptions( $this->name, '' );
		$this->subscriptions = [];
		$caught              = Worker_Should_Stop::attempt_each( $this->siblings, static fn ( Node $sibling ) => $sibling->remove_node() );
		$this->siblings      = [];
		$this->registrations = [];
		$this->set_state     = [];
		$this->sink          = null;
		$this->target        = '';
		$this->patron        = null;
		$this->publisher     = null;
		$this->interpreter   = null;
		if ( '' !== $this->name ) {
			Core::unregister_node( $this->name );
			$this->name = '';
		}
		Worker_Should_Stop::raise( $caught );
	}

	/**
	 * Re-key every subscription from one listener name to another at its
	 * emitter; an empty `$to` only drops them.
	 *
	 * @param string $from The name the emitters hold them under.
	 * @param string $to   The name to hold them under now, or '' to drop.
	 */
	private function move_subscriptions( string $from, string $to ): void {
		foreach ( $this->subscriptions as $emitter => $events ) {
			$node = Core::node( (string) $emitter );
			foreach ( $events as $event => $callback ) {
				$node?->unregister( $event, $from );
				if ( '' !== $to ) {
					$node?->register( $event, $to, $callback );
				}
			}
		}
	}

	/**
	 * Point the node at a target, replacing whatever it held. Tee overrides this
	 * to append, because its target is a fan-out list.
	 *
	 * @param string $target Path to stamp into an empty TO.
	 */
	public function connect_node( string $target ): void {
		$this->target( $target );
	}

	/**
	 * Get/set target — the ROUTING contract. String or array (Tee uses the
	 * array form for fan-out), so an array answer means this node fans out.
	 * Naming a target on a class that declares none is refused before the
	 * field changes; clearing one never is.
	 *
	 * @param string|array<int,string>|null $value New target (null = getter).
	 * @return string|array<int,string>
	 * @throws \RuntimeException When the class declares no target.
	 */
	public function target( $value = null ) {
		if ( null !== $value ) {
			if ( [] !== self::target_list( $value ) ) {
				self::refuse_target_on( static::class, $this->name );
			}
			$this->target = $value;
		}
		return $this->target;
	}

	/**
	 * A target value as a list: the array form as-is, a non-empty scalar
	 * wrapped, an unset scalar dropped.
	 *
	 * @param string|array<int,string> $value Scalar or fan-out target.
	 * @return list<string>
	 */
	public static function target_list( $value ): array {
		return \is_array( $value ) ? \array_values( $value ) : ( '' !== $value ? [ $value ] : [] );
	}

	/**
	 * Refuse a target to a class whose schema declares `has_target: false`:
	 * nothing it does reads one, and the console offers it no out-port. The
	 * two target setters, `Node::target()` and `Fanout_Targets::target()`, ask
	 * here before they change anything, and every `connect_node()` sets through
	 * them; a group asks for its child class, and `Topology_Analyzer` asks for
	 * each line it walks, so a topology a load would refuse never draws the edge.
	 *
	 * @param class-string<Node> $class The class the target would be set on.
	 * @param string             $name  The node's name, for the refusal.
	 * @throws \RuntimeException When the class declares no target.
	 */
	public static function refuse_target_on( string $class, string $name ): void {
		if ( false === ( $class::node_schema()['has_target'] ?? true ) ) {
			throw new \RuntimeException( \esc_html( "{$name} takes no target: " . Command_Interpreter_Node::shell_name_for( $class ) . ' declares has_target false' ) );
		}
	}

	/**
	 * Whether a schema puts its class on the console palette: a category that
	 * is neither empty nor `Hidden`, and no `hidden` flag. Only these may
	 * declare `owns`.
	 *
	 * @param array<string,mixed> $schema A class's `node_schema()`.
	 * @return bool True when the palette lists the class.
	 */
	public static function in_palette( array $schema ): bool {
		$category = $schema['category'] ?? '';
		return '' !== $category && 'Hidden' !== $category && ! self::hidden( $schema );
	}

	/**
	 * Whether a schema keeps its class off every console surface.
	 *
	 * @param array<string,mixed> $schema A class's `node_schema()`.
	 */
	public static function hidden( array $schema ): bool {
		return true === ( $schema['hidden'] ?? false );
	}

	/**
	 * Tear the sibling in a slot down and empty the slot — the exact inverse of
	 * `publish_sibling()`, which owns the name as well as the slot. Forgetting
	 * one without unregistering the other is the bug in both directions: a slot
	 * the four cascades still reach re-registers a dead sibling under the new
	 * name, and a name the registry still holds refuses the slot's next
	 * occupant for the life of the process. A no-op on an empty slot.
	 *
	 * A teardown that throws empties the slot whenever it got as far as tearing
	 * the sibling down — `remove_node()` unregisters and drops the publisher
	 * before raising what its own siblings threw — because a slot still holding
	 * that dead node re-registers it on the next rename and refuses the next
	 * publish as occupied. A sibling that refused before tearing down is still
	 * registered and still names this node its publisher, so it keeps the slot
	 * for a later retraction: dropping it would strand a live, registered node
	 * that is not reached by any cascade.
	 *
	 * @param string $kind The kind the slot was published under.
	 * @throws \Throwable What the sibling's teardown threw.
	 */
	protected function retract_sibling( string $kind ): void {
		$sibling = $this->siblings[ $kind ] ?? null;
		try {
			$sibling?->remove_node();
		} finally {
			if ( $this !== $sibling?->publisher ) {
				unset( $this->siblings[ $kind ] );
			}
		}
	}

	/**
	 * The destinations a node of this class writes without routing through
	 * `target`, as the commands sent to its `:config` leave them once each has
	 * applied in order: the command-side twin of `declared_targets()`, so a command
	 * that undoes another takes its destination back. None by default.
	 *
	 * A Vault_Group declares these for the commands it recorded.
	 * `Topology_Analyzer` asks again after each `command_node` line and draws
	 * a `pair` edge to each destination the line adds, and takes back each
	 * one it drops that the node's arguments do not name, so the runtime and
	 * the drawing judge a command by one rule.
	 *
	 * @param list<array{0:string,1:list<string>}> $commands Each verb and its argument tokens, in the order sent.
	 * @return list<string>
	 */
	public static function command_targets( array $commands ): array {
		return [];
	}

	/**
	 * What `Topology_Analyzer::graph_for()` adds to a node of this class's
	 * row, judged from its `make_node` arguments and the commands sent to its
	 * `:config` in evaluation order, `set_*target` lines aside, which are
	 * edges. Nothing by default; a key the row already carries is kept. A
	 * throw makes the topology unreadable, as a refused `make_node` line does.
	 *
	 * @param list<string>                         $args     The `make_node` arguments, quotes stripped.
	 * @param list<array{0:string,1:list<string>}> $commands Each verb and its argument tokens, in the order sent.
	 * @return array<string,int|string|list<string>|list<array{source:string,target:string}>>
	 */
	public static function graph_facts( array $args, array $commands ): array {
		return [];
	}

	/**
	 * Whether this node takes a `{name}:config` interpreter: true unless its
	 * builder called `without_config()`. A node building siblings of its own
	 * reads it to carry the opt-out to them.
	 */
	public function wants_config(): bool {
		return $this->wants_config;
	}

	/**
	 * The published siblings, keyed by the kind each slot was published under.
	 * A READ of the map the four cascades walk, so a publisher enumerating what
	 * it built keeps no second list in step by hand; `Vault_Group_Node` reads
	 * its members here.
	 *
	 * @return array<array-key,Node> PHP keys an all-digit kind as an int.
	 */
	protected function siblings(): array {
		return $this->siblings;
	}

	/**
	 * Hand a durable read cursor off at a stop. A node holding none has nothing
	 * to hand off; `Durable_Reader` answers it for every node that does, so the
	 * worker's shutdown sweep and a `Vault_Group` retracting a child both key
	 * on the capability rather than on a class.
	 *
	 * @param string $stop_reason             `timeout` or `memory` for a cooperative stop; anything else is operational.
	 * @param bool   $baseline_near_watermark Memory stop only: the fresh baseline already sat near the watermark.
	 */
	public function hand_off_cursor( string $stop_reason = '', bool $baseline_near_watermark = false ): void {}

	/**
	 * The nodes this one stands for as a fan-out target, or null when it stands
	 * for itself. `Fanout_Targets` delivers a target answering a list to each
	 * member instead, so one `connect_node` reaches every node a group keeps;
	 * an empty list is a group with nowhere to deliver yet.
	 *
	 * @return list<Node>|null
	 */
	public function members(): ?array {
		return null;
	}

	/**
	 * Listen at another node under this node's own name, for this node's life.
	 *
	 * The subscription is recorded here and registered at the emitter, which is
	 * resolved by name at every (re)registration: a missing one — no fleet in
	 * REPL or request scope — is a no-op. Unnamed, it registers once named. A
	 * rename moves it and `remove_node()` drops it; re-subscribing the same
	 * emitter and event replaces the callback. A closure has no self-heal of its
	 * own, since `notify()` keeps any listener that does not return exactly
	 * false, so without the move an entry left under the old name keeps firing
	 * while a later node taking that name overwrites it.
	 *
	 * @param string   $emitter  Name of the node whose event this listens to.
	 * @param string   $event    An event the emitter declares.
	 * @param \Closure $callback Takes the payload; returning exactly false unsubscribes at the emitter.
	 */
	protected function subscribe( string $emitter, string $event, \Closure $callback ): void {
		$this->subscriptions[ $emitter ][ $event ] = $callback;
		if ( '' !== $this->name ) {
			Core::node( $emitter )?->register( $event, $this->name, $callback );
		}
	}

	/**
	 * The node that published this one as an owned sibling, or null. A published
	 * sibling is rebuilt by its publisher, so `dump_config` never replays it.
	 */
	public function publisher(): ?Node {
		return $this->publisher;
	}

	/** This node's `:config` interpreter, or null when it declares no commands. */
	public function interpreter(): ?Command_Interpreter_Node {
		return $this->interpreter;
	}

	/**
	 * Quote+escape ONE token so the Shell's tokenizer recovers it exactly.
	 * Mirror of the JS serializeArg; the inverse of tokenize().
	 *
	 * `<` is a metachar here for a reason that is not about tokenizing: a
	 * stored argument may hold an UNEXPANDED `<…>`, which is what a single-quoted
	 * token (`<config:logs_dir>/jobs.'<topology>'`) hands a node. Emitted
	 * bare it would be expanded on the next load, because `interpolate()` runs
	 * before `tokenize()` — so quoting is what preserves the deferral.
	 *
	 * @param string $token The token to quote.
	 * @return string The token, quoted and escaped where it needs to be.
	 */
	public static function serialize_arg( string $token ): string {
		if ( '' !== $token && ! \preg_match( '/[\s\'"`\\\\#;<]/', $token ) ) {
			return $token;
		}
		return "'" . \str_replace( [ '\\', "'" ], [ '\\\\', "\\'" ], $token ) . "'";
	}

	/**
	 * Read the trace level, or set it; a negative level clamps to zero.
	 *
	 * @param int|null $level New level (null = pure getter).
	 * @return int The level now in force.
	 */
	public function debug_state( ?int $level = null ): int {
		if ( null !== $level ) {
			$this->debug_state = \max( 0, $level );
		}
		return $this->debug_state;
	}

	/**
	 * Set/get the sink, cascading a set to every owned sibling so it too sinks into
	 * `_command_interpreter` like any other node — Tachikoma's rule that everything
	 * sinks into the command interpreter. Without the cascade a sibling's emits drop
	 * on the floor.
	 *
	 * @param Node|null $node New sink; omit the argument entirely to read.
	 * @return Node|null The current sink.
	 */
	public function sink( ?Node $node = null ): ?Node {
		if ( \func_num_args() > 0 ) {
			$this->sink = $node;
			foreach ( $this->siblings as $sibling ) {
				$sibling->sink( $node );
			}
		}
		return $this->sink;
	}

	/** Messages this node has filled. */
	public function counter(): int {
		return $this->counter;
	}

	/** Largest single message this node has moved, in bytes. */
	public function largest_msg_sent(): int {
		return $this->largest_msg_sent;
	}

	/** Bytes this node has pulled off a stream. */
	public function bytes_read(): int {
		return $this->bytes_read;
	}

	/** Bytes this node has pushed to a stream. */
	public function bytes_written(): int {
		return $this->bytes_written;
	}

	/**
	 * Drop one listener from an event. A node registered by NAME owes this call in
	 * its own `remove_node()`; short of that, the first `notify()` after teardown
	 * warns and drops it.
	 *
	 * @param string $event    Event name.
	 * @param string $listener Closure ID or Node name.
	 */
	public function unregister( string $event, string $listener ): void {
		unset( $this->registrations[ $event ][ $listener ] );
	}

	/**
	 * The cached `set_state()` payload for an event.
	 *
	 * @param string $event Event name.
	 * @return string|null Null when the event has never been set.
	 */
	public function get_state( string $event ): ?string {
		return $this->set_state[ $event ] ?? null;
	}

	/**
	 * Name-registered listeners keyed by event, which is what `dump_metadata` draws
	 * registration edges from. A closure listener has no name to draw an edge to, so
	 * it is skipped, and an event left with no names is omitted entirely.
	 *
	 * @return array<string,list<string>> Event name => listener Node names.
	 */
	public function registered_listeners(): array {
		$out = [];
		foreach ( $this->registrations as $event => $listeners ) {
			$names = [];
			foreach ( $listeners as $listener => $cb ) {
				if ( null === $cb ) {
					$names[] = $listener;
				}
			}
			if ( [] !== $names ) {
				$out[ $event ] = $names;
			}
		}
		return $out;
	}

	/**
	 * Clear the target. A base node holds exactly one, so any argument clears it;
	 * Tee's fan-out override reads the argument and removes only that entry.
	 *
	 * @param string $target The entry to remove — meaningful only in the override.
	 */
	public function disconnect_node( string $target = '' ): void {
		$this->target = '';
	}

	/**
	 * Hook for a node to contribute extra fields into its dump_metadata row. The
	 * interpreter merges the return with `+=`, so the fixed dump_metadata keys win
	 * and the hook can only add. Base returns nothing.
	 *
	 * @return array<string,mixed>
	 */
	public function dump_metadata(): array {
		return [];
	}

	/**
	 * The arguments of the owned Table `$suffix` that a `make_node` of this
	 * class builds, read by the analyzer without building a node. A class
	 * whose `owns` declares a Table overrides it.
	 *
	 * @param string       $suffix    The owned Table's suffix in `owns`.
	 * @param string       $owner     The owner's name, as written.
	 * @param list<string> $arguments The owner's `make_node` arguments.
	 * @return array{namespace: string, ttl: string, backend: string}
	 * @throws \LogicException Always, on a class declaring no owned Table.
	 */
	public static function owned_table( string $suffix, string $owner, array $arguments ): array {
		throw new \LogicException( \esc_html( static::class . " declares no owned Table {$suffix} for {$owner}" ) );
	}

	/**
	 * Topology console manifest: the palette entry and the node's configuration
	 * form. A subclass overrides it to declare its positional arguments, commands,
	 * registrations, category and description. Under `owns` it names the
	 * siblings the topology analyzer must know it builds, suffix to class; only
	 * a class `in_palette()` may declare them. The empty category the base
	 * returns keeps a node out of the palette and the class catalog, which is
	 * what a pure plumbing node wants; a `Hidden` category keeps it off the
	 * palette alone, and a `hidden` flag off every console surface.
	 *
	 * @return array<string,mixed>
	 */
	public static function node_schema(): array {
		return [
			'category'      => '',
			'description'   => '',
			'arguments'     => [],
			'commands'      => [],
			'registrations' => [],
			'accepts_fill'  => true,
			'has_target'    => true,
			'owns'          => [],
		];
	}
}