package dev.ltms.bridged.member;

import dev.ltms.bridged.config.BridgedConfig;
import dev.ltms.bridged.herdr.Agent;
import dev.ltms.bridged.herdr.AgentControl;
import dev.ltms.bridged.herdr.HerdrException;
import dev.ltms.bridged.herdr.Tab;
import dev.ltms.bridged.herdr.Workspace;
import dev.ltms.bridged.herdr.WorkspaceControl;
import dev.ltms.bridged.peer.Capability;
import dev.ltms.bridged.peer.CharterReceipt;
import dev.ltms.bridged.peer.MemberRole;
import dev.ltms.bridged.peer.PeerHandle;
import dev.ltms.bridged.peer.PeerLauncher;
import dev.ltms.bridged.peer.PeerUnreachableException;
import dev.ltms.bridged.peer.SpawnRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.security.SecureRandom;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Function;
import java.util.function.LongSupplier;
import java.util.function.Supplier;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * Abstract base for {@link PeerLauncher} adapters that materialize a peer as a <em>herdr</em>
 * agent (a CLI coding agent running in a herdr tab/pane). It owns everything that is the same
 * regardless of <em>which</em> coding agent runs: tab/pane placement, the CB-306 spawn-readiness
 * gate, unique naming, CB-117 orphan reap, teardown, {@link #list() listing}, and cwd resolution.
 *
 * <p>Two seams are peer-specific and supplied by the concrete adapter:
 * <ul>
 *   <li>{@code namePrefix} (constructor arg) — the label prefix ({@code claude}, {@code opencode})
 *       that drives both unique naming and the orphan-reap pattern, so each adapter reaps only its
 *       own kind of pane and never another's.</li>
 *   <li>{@link #buildLaunch(BridgedConfig.Profile, LaunchSpec)} — the peer-specific env map + argv, including any
 *       subscription/guard check, MCP mount, and instruction injection. The base never sees how the
 *       peer is configured; it only places and starts the returned {@link Launch}.</li>
 * </ul>
 *
 * <p>Placement: in the default {@code tab} policy a peer lands in its own tab inside a dedicated
 * worker space (found-or-created once, then shared), so peers never split or clutter the user's
 * real work spaces. Teardown removes the peer's pane <em>and</em> its now-empty tab, tolerating an
 * already-gone peer so a repeated DELETE is harmless.
 */
public abstract class HerdrPeerLauncher implements PeerLauncher {

    private static final Logger log = LoggerFactory.getLogger(HerdrPeerLauncher.class);

    /** herdr rejects a duplicate agent {@code name}; we retry a bumped name this many times. */
    private static final int NAME_RETRIES = 8;

    /**
     * Retries for {@code agent.start} against a seed pane whose shell has not reached its prompt
     * yet — {@code tab.create}/{@code pane.split} return as soon as the pane exists, and herdr
     * refuses to start an agent in a pane that is not "an available shell" ({@code agent_pane_busy}).
     */
    private static final int SHELL_READY_RETRIES = 20;

    private final String namePrefix;                       // label prefix: naming + reap scheme
    private final AgentControl agents;
    private final WorkspaceControl spaces;
    private final Map<String, BridgedConfig.Profile> profiles; // profile name → spawn settings
    private final String defaultProfile;                      // profile a no-arg spawn uses (nullable)

    /** Host env lookup (injectable for tests); adapters read it in {@link #buildLaunch}. */
    protected final Function<String, String> env;

    private final AtomicLong nameSeq = new AtomicLong(); // per-peer counter (herdr agent names only)

    /**
     * Live fleet config, read once per spawn. A null supplier or value leaves tab labels at their
     * default and supplies no role charter. A profile's own {@code tabLabel} still overrides it.
     *
     * <p>CB-559: a supplier rather than a snapshot, so a config reload affects the next launch
     * without a restart. Existing tabs keep the label they were given.
     */
    private final Supplier<BridgedConfig.Fleet> fleet;

    /** The final instruction always requires a bridge reply when the bridge MCP is mounted. */
    protected static final String REPLY_CHARTER =
            "You are a spawned member in the claude-bridge fleet. Every message you receive arrives "
            + "through the bridge, and the ONLY channel back to the sender is the bridge_reply MCP tool. "
            + "Text you write in your terminal is NOT sent anywhere — the sender cannot see your screen, "
            + "so an in-terminal answer is silently discarded. Therefore you MUST end EVERY turn by calling "
            + "bridge_reply with `content` set to your complete response. This holds for every message without "
            + "exception — tasks, questions, clarifications, acknowledgements, and ordinary back-and-forth "
            + "conversation. Call bridge_reply exactly once, as the final action of your turn, with your full "
            + "answer in `content`; never wait for confirmation first. If you end a turn without calling "
            + "bridge_reply, the sender receives nothing and the exchange stalls.";

    /**
     * Tab numbers, counted per {@code role/profile} pair (CB-557).
     *
     * <p>Deliberately not {@link #nameSeq}. That counter is shared by every profile this launcher
     * serves, because its job is to make herdr <em>agent names</em> unique. Reusing it for the tab
     * label made the numbers global, so sibling tabs read {@code #4}, {@code #9}, {@code #17} — gaps
     * that look like a member died. Counting per role+profile makes {@code dev: sonnet #2} mean the
     * second sonnet dev, which is what a reader assumes it means.
     *
     * <p>Resets when the daemon restarts, and that is fine: the label is a human-facing hint, not an
     * identity. Identity is {@link PeerHandle#id()}.
     */
    private final ConcurrentMap<String, AtomicLong> labelSeq = new ConcurrentHashMap<>();

    private final long spawnReadyTimeoutMs; // 0 = disable gate (legacy non-blocking spawn)
    private final LongSupplier nowMillis; // monotonic clock (injectable for tests)
    private final Runnable sleeper; // sleep/wait hook (injectable for tests; never real-sleep in unit tests)

    // Per-process token mixed into each peer name so a fresh process (nameSeq back at 0) cannot
    // collide with same-profile peers that outlived a restart. See startUniquelyNamed.
    private final String nameNonce = String.format("%06x", new SecureRandom().nextInt(1 << 24));

    // CB-519: PeerHandle.id() is a host-unique opaque UUID, decoupled from the herdr pane id. The
    // routing/registry key is the UUID; the herdr pane id is a launcher-private placement/teardown
    // coordinate. This map bridges the two so stop(id) can resolve a host-unique key back to the
    // exact pane it must tear down. The pane id is launcher-private (never the routing key) — see
    // PeerHandle.id().
    private final ConcurrentMap<String, String> paneByAgentId = new ConcurrentHashMap<>();
    private final AtomicBoolean resetUnsupportedLogged = new AtomicBoolean();

    /**
     * @param namePrefix          label prefix for this peer kind (drives naming and reap)
     * @param agents              herdr agent control (start, status, close)
     * @param spaces              workspace / tab control (ensure, create, close)
     * @param profiles            configured peer profiles
     * @param defaultProfile      profile a no-argument spawn uses (nullable)
     * @param env                 host env lookup (injectable for tests)
     * @param spawnReadyTimeoutMs max ms to wait for injectable state (0 disables the gate)
     * @param nowMillis           monotonic clock source (e.g. {@code System::currentTimeMillis})
     * @param sleeper             sleep/wait hook (never called when the gate is disabled); the poll
     *                            interval is baked into this hook, so the base needs no poll field
     */
    protected HerdrPeerLauncher(String namePrefix, AgentControl agents, WorkspaceControl spaces,
                                Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
                                Function<String, String> env,
                                long spawnReadyTimeoutMs,
                                LongSupplier nowMillis, Runnable sleeper) {
        this(namePrefix, agents, spaces, profiles, defaultProfile, env, spawnReadyTimeoutMs,
                nowMillis, sleeper, null);
    }

    /**
     * As above, plus the live {@code fleet} config (CB-557).
     *
     * @param fleet live fleet config, read once per spawn; {@code null} ⇒ default tab label and no
     *              role charter. A separate constructor rather than a new parameter on the one
     *              above, so every existing call site keeps the default without an edit.
     */
    protected HerdrPeerLauncher(String namePrefix, AgentControl agents, WorkspaceControl spaces,
                                Map<String, BridgedConfig.Profile> profiles, String defaultProfile,
                                Function<String, String> env,
                                long spawnReadyTimeoutMs,
                                LongSupplier nowMillis, Runnable sleeper,
                                Supplier<BridgedConfig.Fleet> fleet) {
        this.fleet = fleet;
        this.namePrefix = namePrefix;
        this.agents = agents;
        this.spaces = spaces;
        this.profiles = Map.copyOf(profiles);
        this.defaultProfile = defaultProfile;
        this.env = env;
        this.spawnReadyTimeoutMs = spawnReadyTimeoutMs;
        this.nowMillis = nowMillis;
        this.sleeper = sleeper;
    }

    // --- adapter seams -------------------------------------------------------------------------

    /**
     * Build the peer-specific launch for {@code cfg}: the environment map and argv handed to herdr.
     * Any subscription/guard check, MCP mount, and instruction injection happen here. The env map
     * and argv are adapter-private; the base only places and starts what is returned.
     */
    protected abstract Launch buildLaunch(BridgedConfig.Profile cfg, LaunchSpec spec);

    /** Direct transport access for peer-specific, non-turn control operations. */
    protected final AgentControl agents() {
        return agents;
    }

    /** Resolve the public peer id to the launcher's private herdr target. */
    protected final String agentTarget(String id) {
        return paneByAgentId.get(id);
    }

    @Override
    public boolean clearContext(String id) {
        if (resetUnsupportedLogged.compareAndSet(false, true)) {
            log.warn("context reset is unsupported for peer kind {}; clearAfterTurn is a no-op",
                    namePrefix);
        }
        return false;
    }

    /**
     * A peer-specific launch: the herdr {@code env} map and {@code argv}, plus — for an adapter
     * that carries durable session identity (CB-547a) — the peer's OWN session id
     * ({@link PeerHandle#agentSessionId()}), known before the peer has written anything. Null for
     * a launch that carries no identity.
     */
    protected record Launch(Map<String, String> env, List<String> argv, String agentSessionId) {

        /** A launch without a discoverable agent session id (an adapter that carries none). */
        Launch(Map<String, String> env, List<String> argv) {
            this(env, argv, null);
        }
    }

    /** All per-spawn values adapters may need, including the base-composed effective charter. */
    protected record LaunchSpec(String sessionName, String resumeSessionId, MemberRole role, String charter) {
    }

    // --- profile surface -----------------------------------------------------------------------

    /** The configured peer profile names (what {@code spawn(profile)} accepts). */
    @Override
    public Set<String> profiles() {
        return profiles.keySet();
    }

    /** The parity-overlay file list for {@code profileName} (default list when unset). */
    @Override
    public List<String> parityOverlay(String profileName) {
        String name = (profileName == null || profileName.isBlank()) ? defaultProfile : profileName;
        if (name == null || name.isBlank()) {
            return List.of();
        }
        BridgedConfig.Profile cfg = profiles.get(name);
        return cfg == null ? List.of() : cfg.parityOverlay();
    }

    /** The profile a no-argument spawn uses, or {@code null} if none is configured. */
    @Override
    public String defaultProfile() {
        return defaultProfile;
    }

    /**
     * {@inheritDoc}
     *
     * <p>One {@link HerdrPeerLauncher} instance always serves exactly one adapter kind, so every
     * profile it owns shares that adapter's {@link #capabilities()} — {@code profileName} only
     * needs validating (throwing on an unknown profile, same as {@link #spawn}), not routing.
     */
    @Override
    public Set<Capability> capabilitiesFor(String profileName) {
        requireProfile(profileName);
        return capabilities();
    }

    /** The configured profiles, for adapter capability decisions (e.g. any git-token grant). */
    protected Collection<BridgedConfig.Profile> profileConfigs() {
        return profiles.values();
    }

    /** Resolve {@code profileName} (null/blank → default) to its config, or throw with the options. */
    protected BridgedConfig.Profile requireProfile(String profileName) {
        String name = (profileName == null || profileName.isBlank()) ? defaultProfile : profileName;
        if (name == null || name.isBlank()) {
            throw new IllegalArgumentException("no default worker profile is configured — "
                    + "pass a profile; configured: " + profiles.keySet());
        }
        BridgedConfig.Profile cfg = profiles.get(name);
        if (cfg == null) {
            throw new IllegalArgumentException("unknown worker profile '" + name
                    + "' — configured: " + profiles.keySet());
        }
        return cfg;
    }

    // --- spawn ---------------------------------------------------------------------------------

    /**
     * A started peer plus the launch's agent-session id (the resume handle, or null) and the
     * charter receipt (CB-571) the base composed for it.
     */
    private record Spawned(Agent agent, String agentSessionId, CharterReceipt receipt) {
    }

    /**
     * Spawn a peer. {@code profileName} null/blank → the default profile. The working directory
     * (CB-112) is resolved by {@link #resolveCwd}: an explicit {@code requestedCwd}, else the
     * profile's configured {@code cwd}, else {@code callerCwd} (the primary's cwd, when the spawn
     * came from the primary over MCP), else the daemon's cwd — never assumed to be {@code $HOME}.
     * The adapter's {@link #buildLaunch} runs before any herdr call.
     */
    protected Agent spawnInternal(String profileName, String requestedCwd, String callerCwd) {
        return spawnInternal(profileName, requestedCwd, callerCwd, null, null).agent();
    }

    /** Pre-CB-557 shape: no explicit role, so the tab is labelled as a {@code dev}. */
    protected Spawned spawnInternal(String profileName, String requestedCwd, String callerCwd,
                                    String sessionName, String resumeSessionId) {
        return spawnInternal(profileName, requestedCwd, callerCwd, sessionName, resumeSessionId,
                MemberRole.DEV);
    }

    /**
     * Spawn a peer with session identity (CB-547a). The session values, role, and charter are
     * threaded from the {@link SpawnRequest} into {@link #buildLaunch(BridgedConfig.Profile,
     * LaunchSpec)}, and the launch's resolved agent-session id is returned alongside the agent so
     * the caller can put it on the {@link PeerHandle}.
     */
    protected Spawned spawnInternal(String profileName, String requestedCwd, String callerCwd,
                                    String sessionName, String resumeSessionId, MemberRole role) {
        BridgedConfig.Profile cfg = requireProfile(profileName);
        BridgedConfig.Fleet liveFleet = fleet == null ? null : fleet.get();
        String roleCharter = liveFleet == null ? null : liveFleet.charterFor(role);
        String replyCharter = cfg.hasMcp() ? REPLY_CHARTER : null;
        String charter = roleCharter == null ? replyCharter
                : replyCharter == null ? roleCharter : roleCharter + "\n\n" + replyCharter;
        // CB-571: fingerprint the exact composed charter bytes once, here in the base, before the
        // string leaves for an adapter — so Claude and OpenCode derive the same digest. A failed
        // start has no bridge_spawn result and no roster row, so the failure log below is the only
        // surface the byte count can appear on. The charter text itself is never logged.
        CharterReceipt receipt = CharterReceipt.compose(role, cfg.profile(), roleCharter, charter);
        try {
            Launch launch = buildLaunch(cfg, new LaunchSpec(sessionName, resumeSessionId, role, charter));
            String cwd = resolveCwd(requestedCwd, cfg, callerCwd);
            Agent agent = cfg.tabPlacement()
                    ? spawnInTab(cfg, launch.env(), launch.argv(), cwd, role, liveFleet)
                    : spawnAsPane(cfg, launch.env(), launch.argv(), cwd, charter);
            logCharterReceipt(receipt, true);
            return new Spawned(agent, launch.agentSessionId(), receipt);
        } catch (RuntimeException e) {
            logCharterReceipt(receipt, false);
            throw e;
        }
    }

    /**
     * The one place the charter's size and digest appear in the logs. {@code success} true after a
     * start, false from the failure path of {@link #spawnInternal} where no handle or roster row
     * exists to carry the receipt. Always metadata only — never the charter text.
     */
    private static void logCharterReceipt(CharterReceipt receipt, boolean success) {
        String role = receipt.role() == null ? "" : receipt.role().wireName();
        if (success) {
            log.info("spawned role={} profile={} charterSource={} charterSha256={} charterBytes={}",
                    role, receipt.profile(), receipt.charterSource(),
                    receipt.charterSha256(), receipt.charterBytes());
        } else {
            log.warn("spawn failed; charter role={} profile={} charterSource={} charterSha256={} charterBytes={}",
                    role, receipt.profile(), receipt.charterSource(),
                    receipt.charterSha256(), receipt.charterBytes());
        }
    }

    /**
     * The next tab number for {@code role} on {@code profile}, starting at 1.
     *
     * <p>Starts at 1 rather than 0 because the number is read by a person: {@code "dev: sonnet #1"}
     * is the first one, and {@code #0} invites the question of where {@code #1} went.
     */
    private long nextLabelSeq(MemberRole role, String profile) {
        String key = (role == null ? "" : role.wireName()) + "/" + profile;
        return labelSeq.computeIfAbsent(key, _ -> new AtomicLong()).incrementAndGet();
    }

    /**
     * {@inheritDoc}
     *
     * <p>Delegates to {@link #spawnInternal} and wraps the resulting herdr {@link Agent} in a
     * {@link WorkerHandle} whose {@link PeerHandle#id()} is a fresh <em>host-unique</em> opaque
     * UUID (CB-519), deliberately decoupled from the herdr pane id: the id is the registry/routing
     * key and must never collide across daemon processes on the same host, while the herdr pane id
     * stays a launcher-private placement/teardown coordinate, remembered here so {@link #stop}
     * can resolve the host-unique key back to its pane. When {@code spawnReadyTimeoutMs > 0},
     * blocks until the peer's herdr status is injectable or the timeout elapses; on timeout the
     * pane is closed (no orphan) and a {@link PeerUnreachableException} is thrown.
     */
    @Override
    public PeerHandle spawn(SpawnRequest req) {
        Spawned spawned = spawnInternal(req.profileName(), req.requestedCwd(), req.callerCwd(),
                req.sessionName(), req.resumeSessionId(), req.role());
        Agent agent = spawned.agent();
        String paneId = agent.paneId();
        if (spawnReadyTimeoutMs > 0) {
            waitUntilInjectableOrThrow(paneId);
        }
        // CB-519: the handle id is a host-unique UUID; the herdr pane it maps to stays internal.
        String id = UUID.randomUUID().toString();
        paneByAgentId.put(id, paneId);
        return new WorkerHandle(id, agent.terminalId(), requireProfile(req.profileName()).profile(),
                req.sessionName(), spawned.agentSessionId(), spawned.receipt());
    }

    @Override
    public String effectiveCwd(SpawnRequest req) {
        return effectiveCwd(req.profileName(), req.requestedCwd(), req.callerCwd());
    }

    /**
     * CB-301: the effective working directory a spawn for {@code profileName} would use, without
     * actually spawning.
     */
    private String effectiveCwd(String profileName, String requestedCwd, String callerCwd) {
        return resolveCwd(requestedCwd, requireProfile(profileName), callerCwd);
    }

    /**
     * CB-112 cwd resolution: spawn arg → profile config → the primary's cwd → the daemon's cwd.
     * Never returns {@code null}/blank: {@code "."} (the daemon's own working directory) is the
     * guaranteed last resort so a pathological environment with an unset {@code user.dir} still
     * honours the "never assume {@code $HOME}" contract rather than letting herdr default the pane.
     */
    private static String resolveCwd(String requestedCwd, BridgedConfig.Profile cfg, String callerCwd) {
        return firstNonBlank(requestedCwd, cfg.cwd(), callerCwd, System.getProperty("user.dir"), ".");
    }

    private static String firstNonBlank(String... values) {
        for (String v : values) {
            if (v != null && !v.isBlank()) return v;
        }
        return null;
    }

    /** Dedicated worker space → own tab (carrying cwd+env) → start the peer into the seed pane. */
    private Agent spawnInTab(BridgedConfig.Profile cfg, Map<String, String> workerEnv,
                             List<String> argv, String cwd, MemberRole role, BridgedConfig.Fleet liveFleet) {
        Workspace space = spaces.ensureWorkspace(cfg.workspace());
        Tab.Created tab = spaces.createTab(space.workspaceId(), cwd, workerEnv);
        log.info("spawning {} profile={} space={} tab={} cwd={}",
                namePrefix, cfg.profile(), space.workspaceId(), tab.tab().tabId(), cwd);

        Started started;
        try {
            if (tab.rootPaneId() == null) {
                // Protocol 19 starts the agent INTO the seed pane — without one there is nowhere
                // to start, and a partial tab would be left behind.
                throw new IllegalStateException("tab " + tab.tab().tabId()
                        + " had no seed pane in the create response — cannot start a peer in it");
            }
            started = startUniquelyNamed(cfg, argv, tab.rootPaneId());
        } catch (RuntimeException e) {
            // The peer never started — don't leave the tab we just created orphaned.
            // Best-effort cleanup; never let it mask the real spawn failure.
            try {
                spaces.closeTab(tab.tab().tabId());
            } catch (RuntimeException cleanup) {
                log.warn("failed to close orphaned tab {} after spawn error: {}",
                        tab.tab().tabId(), cleanup.getMessage());
            }
            throw e;
        }

        // The peer is LIVE now, in the seed pane itself (no shell pane to drop — protocol 19).
        // Labelling is cosmetic: it must not fail the spawn or orphan the running peer — on error
        // we log and still return it so the caller gets its paneId and can tear it down.
        tidy("label tab " + tab.tab().tabId(),
                () -> spaces.renameTab(tab.tab().tabId(),
                        cfg.renderTabLabel(
                                liveFleet == null ? null : liveFleet.tabLabel(),
                                role, nextLabelSeq(role, cfg.profile()))));
        log.info("{} started pane={} tab={} terminal={}",
                namePrefix, started.agent().paneId(), started.agent().tabId(), started.agent().terminalId());
        return started.agent();
    }

    /** Run a best-effort post-start cleanup step, logging (not throwing) on failure. */
    private void tidy(String what, Runnable step) {
        try {
            step.run();
        } catch (RuntimeException e) {
            log.warn("post-start step failed ({}) — peer is running regardless: {}", what, e.getMessage());
        }
    }

    /**
     * Legacy placement: split the currently-focused tab; the peer still starts in {@code cwd}.
     *
     * <p>CB-571: this is the one legacy log that printed the full argv, and the charter travels
     * inside argv — so the charter text went to the daemon log on every pane-placement spawn. The
     * {@code spawnInTab} path never logs argv, so only this site is fixed. {@code charter} is the
     * composed charter, if any; its argv element is replaced by its digest so the log still shows
     * which args were passed without exposing the charter prose.
     */
    private Agent spawnAsPane(BridgedConfig.Profile cfg, Map<String, String> workerEnv,
                              List<String> argv, String cwd, String charter) {
        log.info("spawning {} (pane placement) profile={} cwd={} argv={}",
                namePrefix, cfg.profile(), cwd, redactCharter(argv, charter));
        String paneId = spaces.splitPane(cwd, workerEnv);
        if (paneId == null) {
            throw new IllegalStateException("pane.split returned no pane — cannot start a peer");
        }
        Agent peer = startUniquelyNamed(cfg, argv, paneId).agent();
        log.info("{} started pane={} terminal={}", namePrefix, peer.paneId(), peer.terminalId());
        return peer;
    }

    /**
     * A copy of {@code argv} with an element equal to {@code charter} replaced by its digest, so
     * the pane log never prints the charter prose. The charter is handed to an adapter as one argv
     * element, so exact-equality is the right match; every other argument passes through unchanged.
     */
    private static List<String> redactCharter(List<String> argv, String charter) {
        if (charter == null || charter.isBlank() || argv == null || argv.isEmpty()) {
            return argv;
        }
        String digest = CharterReceipt.digestOf(charter);
        return argv.stream()
                .map(a -> a.equals(charter) ? "<charter sha256=" + digest + ">" : a)
                .toList();
    }

    /** A started peer together with the sequence its unique name/label used. */
    private record Started(Agent agent, long seq) {
    }

    /**
     * Start the peer under a unique herdr agent name. herdr requires each running agent's
     * {@code name} to be distinct (a 2nd identical {@code name} fails {@code agent_name_taken}) —
     * the exact case that makes multiple peers useful. The name is
     * {@code <prefix>-<profile>-<nonce>-<seq>}: {@code seq} distinguishes peers within this process,
     * and the per-process {@code nonce} keeps a fresh process (whose {@code seq} restarts at 0) from
     * colliding with same-profile peers that outlived a restart. The retry is a belt-and-braces
     * backstop for the astronomically unlikely nonce+seq clash; the name is a label only — herdr
     * detects kind and status from terminal output, not from it.
     */
    private Started startUniquelyNamed(BridgedConfig.Profile cfg, List<String> argv, String paneId) {
        // Protocol 19 resolves the executable from the agent kind (== namePrefix here), so
        // argv[0] — the configured executable — is dropped and only the extra args are passed.
        List<String> args = argv.isEmpty() ? argv : argv.subList(1, argv.size());
        HerdrException last = null;
        for (int attempt = 0; attempt < NAME_RETRIES; attempt++) {
            long seq = nameSeq.incrementAndGet();
            String name = namePrefix + "-" + cfg.profile() + "-" + nameNonce + "-" + seq;
            try {
                return new Started(startAwaitingShellPrompt(name, args, paneId), seq);
            } catch (HerdrException e) {
                if (!"agent_name_taken".equals(e.code())) throw e;
                log.debug("peer name '{}' taken, retrying", name);
                last = e;
            }
        }
        throw last;
    }

    /** Start the agent into {@code paneId}, waiting out the seed shell's boot with the sleeper. */
    private Agent startAwaitingShellPrompt(String name, List<String> args, String paneId) {
        HerdrException busy = null;
        for (int attempt = 0; attempt < SHELL_READY_RETRIES; attempt++) {
            try {
                return agents.start(name, namePrefix, args, paneId);
            } catch (HerdrException e) {
                if (!"agent_pane_busy".equals(e.code())) throw e;
                log.debug("pane {} not at its shell prompt yet, retrying agent.start", paneId);
                busy = e;
                sleeper.run();
            }
        }
        throw busy;
    }

    // --- discovery + reap ----------------------------------------------------------------------

    /** All herdr-tracked agents — discovery for "what peers exist". */
    @Override
    public List<Agent> list() {
        return agents.list();
    }

    /**
     * Reap peer panes left behind by an earlier daemon process (CB-117). herdr keeps a peer's pane
     * alive across a daemon restart <em>by design</em>, and that pane's id is held only by its
     * spawner — so a peer whose owning process exited before issuing the matching teardown leaks
     * with nothing tracking it. On boot we scan herdr for agents whose name matches our
     * {@code <prefix>-<profile>-<nonce>-<seq>} scheme with a nonce <em>other</em> than this
     * process's {@link #nameNonce}, and tear each one down (its pane and, via {@link #stop}, its
     * now-empty dedicated tab). A current-nonce peer is ours and live, so it is left running; a
     * user's own session carries no such name and is never touched. A peer from a <em>different</em>
     * adapter (different prefix) is likewise never touched. Best-effort: a failed listing, or a
     * failure to stop any one peer, is logged and never aborts startup.
     *
     * @return the number of orphaned peers reaped
     */
    @Override
    public int reapOrphanWorkers() {
        List<Agent> all;
        try {
            all = agents.list();
        } catch (RuntimeException e) {
            log.warn("orphan-peer reap skipped — agent.list failed: {}", e.getMessage());
            return 0;
        }
        int reaped = 0;
        for (Agent a : all) {
            if (!isForeignWorker(namePrefix, a.name(), nameNonce)) continue;
            try {
                stop(a.paneId());
                reaped++;
                log.info("reaped orphan {} {} (pane={} tab={}) left by a prior daemon",
                        namePrefix, a.name(), a.paneId(), a.tabId());
            } catch (RuntimeException e) {
                log.warn("could not reap orphan {} {} (pane={}): {}",
                        namePrefix, a.name(), a.paneId(), e.getMessage());
            }
        }
        if (reaped > 0) {
            log.info("orphan-peer reap complete — {} stale {} peer(s) removed at startup", reaped, namePrefix);
        }
        return reaped;
    }

    /** The {@code <prefix>-<profile>-<nonce>-<seq>} name pattern; group 1 captures the 6-hex nonce. */
    static Pattern workerNamePattern(String prefix) {
        return Pattern.compile(prefix + "-.*-([0-9a-f]{6})-\\d+");
    }

    /**
     * Whether {@code name} is a peer of kind {@code prefix} started by a <em>different</em> process
     * than {@code currentNonce} — the reap predicate (CB-117). True only for the prefix's naming
     * scheme with a foreign nonce: a non-peer name, a different adapter's name, or our own live
     * nonce is excluded. Pure and package-private so the decision is unit-testable without herdr.
     */
    static boolean isForeignWorker(String prefix, String name, String currentNonce) {
        String nonce = workerNonce(prefix, name);
        return nonce != null && !nonce.equals(currentNonce);
    }

    /** The 6-hex nonce embedded in a {@code prefix} peer name, or {@code null} if not one. */
    static String workerNonce(String prefix, String name) {
        if (name == null) return null;
        Matcher m = workerNamePattern(prefix).matcher(name);
        return m.matches() ? m.group(1) : null;
    }

    /** This process's peer-name nonce (a label component only; exposed for reaper tests). */
    String nameNonce() {
        return nameNonce;
    }

    // --- teardown ------------------------------------------------------------------------------

    /**
     * Tear a peer down: close the pane, and close its tab <em>only</em> when the peer is that tab's
     * sole occupant. The single-pane check is what makes this safe regardless of how the peer was
     * placed (or a placement-config change across a restart): a pane-placement peer sitting in one
     * of the user's shared tabs has siblings, so its tab is never closed — we only ever remove a
     * tab we created to hold one peer.
     *
     * <p>{@code idOrPane} is the {@link PeerHandle#id()} of a peer this launcher spawned (CB-519's
     * host-unique opaque UUID), resolved through {@link #paneByAgentId} to the pane it must tear
     * down. An argument that is not one of our ids is treated as a raw herdr pane id — the
     * {@link #reapOrphanWorkers() orphan-reap} and spawn-gate-timeout paths, plus any caller that
     * passes a pane directly, keep working without an owning id.
     *
     * <p>Resolves the tab from the pane <em>before</em> closing it. An already-gone pane/tab
     * (repeated DELETE, crashed peer) is treated as success; any other failure propagates so a
     * genuinely failed teardown is not reported as done.
     */
    @Override
    public void stop(String idOrPane) {
        // Teardown knows only the pane, not which profile spawned it. Attempt tab cleanup when any
        // profile uses tab placement (so the bridge may have created a dedicated peer tab); the
        // single-occupant check below is what actually protects the user's shared tabs.
        String paneId = paneByAgentId.remove(idOrPane);
        if (paneId == null) {
            paneId = idOrPane; // raw-pane fallback (reap, gate timeout, pane-addressed callers)
        }
        WorkspaceControl.PaneLocation loc = usesTabPlacement() ? spaces.locatePane(paneId) : null;
        try {
            agents.close(paneId);
        } catch (HerdrException e) {
            if (!isAlreadyGone(e)) throw e;
            log.debug("pane.close({}) ignored — already gone: {}", paneId, e.getMessage());
        }
        if (loc != null && loc.tabPaneCount() == 1) {
            spaces.closeTab(loc.tabId());
        } else if (loc != null) {
            log.debug("not closing tab {} — it holds {} panes (not a dedicated peer tab)",
                    loc.tabId(), loc.tabPaneCount());
        }
    }

    /** Whether any configured profile places peers in their own tab (so tabs may need cleanup). */
    private boolean usesTabPlacement() {
        return profiles.values().stream().anyMatch(BridgedConfig.Profile::tabPlacement);
    }

    /** True when a herdr error means the target is already gone (safe to treat as done). */
    private static boolean isAlreadyGone(HerdrException e) {
        return e.code() != null && e.code().endsWith("_not_found");
    }

    // --- spawn-readiness gate (CB-306) ---------------------------------------------------------

    /**
     * Poll {@link AgentControl#status} until the pane reports an injectable state or the configured
     * timeout elapses. On timeout, close the pane (self-reap) and throw.
     */
    private void waitUntilInjectableOrThrow(String paneId) {
        long deadline = nowMillis.getAsLong() + spawnReadyTimeoutMs;
        while (nowMillis.getAsLong() < deadline) {
            if (agents.status(paneId).injectable()) {
                log.debug("peer pane={} reached injectable state", paneId);
                return;
            }
            sleeper.run();
        }
        log.warn("peer pane={} did not become injectable within {}ms — closing", paneId, spawnReadyTimeoutMs);
        stop(paneId);
        throw new PeerUnreachableException(
                "worker pane " + paneId + " did not reach injectable state within "
                        + spawnReadyTimeoutMs + "ms");
    }

    /**
     * A concrete {@link PeerHandle} wrapping herdr agent coordinates, the profile that spawned it,
     * the session identity the launch resolved (CB-547a): the bridge's logical name and the peer's
     * own session id, both null when the spawn carried no identity — and the charter receipt
     * (CB-571) the base computed for this launch.
     */
    private record WorkerHandle(String id, String terminalId, String profile,
                                String sessionName, String agentSessionId,
                                CharterReceipt receipt) implements PeerHandle {

        @Override
        public CharterReceipt charterReceipt() {
            return receipt;
        }
    }

    // --- shared helpers ------------------------------------------------------------------------

    /** Put {@code k → v} only when {@code v} is present (non-null, non-blank). */
    protected static void putIfPresent(Map<String, String> m, String k, String v) {
        if (v != null && !v.isBlank()) {
            m.put(k, v);
        }
    }

    /** Host env lookup that tolerates an unconfigured (null/blank) var name — returns null then. */
    protected String resolveEnv(String name) {
        return (name == null || name.isBlank()) ? null : env.apply(name);
    }

    /**
     * The parity-neutral git-forge token grant (CB-302): when {@code cfg} opts in via
     * {@code gitTokenEnv} and the token resolves, inject {@code GITEA_TOKEN} plus its paired
     * {@code GITEA_HOST}. Push over SSH is unaffected; the only incremental grant is PR-create.
     * Peer-neutral, so every herdr adapter reuses it unchanged.
     */
    protected void applyGitToken(Map<String, String> workerEnv, BridgedConfig.Profile cfg) {
        if (!cfg.hasGitToken()) {
            return;
        }
        String gitToken = resolveEnv(cfg.gitTokenEnv());
        if (gitToken != null) {
            workerEnv.put("GITEA_TOKEN", gitToken);
            putIfPresent(workerEnv, "GITEA_HOST", resolveEnv(cfg.gitHostEnv()));
        }
    }

    /**
     * CB-592: overlay value that shadows the admin {@code GITEA_ACCESS_TOKEN} a herdr pane
     * otherwise inherits from herdr's own login-shell process environment (gitea issue #77).
     * herdr spawns a pane from its <em>own</em> process environment and layers our map on top —
     * {@link dev.ltms.bridged.herdr.WorkspaceControl#createTab} and {@code #splitPane} send only
     * the keys we put in that map, so any key we never mention passes straight through from
     * herdr's own shell, admin token included.
     *
     * <p>Deliberately a non-blank sentinel, not {@code ""}. Whether an empty-string overlay value
     * overrides an inherited variable or is skipped as blank could not be settled by reading this
     * codebase — herdr's server-side merge is an external process, not something in this repo.
     * A non-blank replacement sidesteps that ambiguity entirely: {@link #baseEnv}'s own {@code
     * PATH} seeding already depends on the overlay reliably replacing an inherited value (see its
     * javadoc), and that is only demonstrated for a non-blank value, so this reuses the same,
     * proven-reliable shape rather than the unverified one.
     */
    private static final String BLOCKED_GITEA_ACCESS_TOKEN =
            "blocked-by-bridged-cb592-see-gitea-issue-77";

    /**
     * Seed a worker's environment (CB-511): the daemon's own {@code PATH}, then the profile's
     * {@code env:} entries, then the CB-592 admin-token shadow.
     *
     * <p>Why this exists: bridged passes herdr an explicit env map, and herdr merges it into
     * <em>its own</em> process environment. So before this, a worker inherited whatever PATH the
     * herdr server happened to be started with — on this host, one from weeks earlier with no JDK
     * and no Maven, which left workers unable to run the build they were being asked to run. The
     * worker's toolchain must follow from configuration, not from how a long-lived daemon was
     * launched.
     *
     * <p>Adapter-specific variables are layered on top of this by {@code buildLaunch} and therefore
     * win. That ordering is deliberate and load-bearing: it stops a profile's {@code env:} from
     * overriding {@code ANTHROPIC_BASE_URL} and slipping past {@link
     * dev.ltms.bridged.guard.SubscriptionGuard}, which is checked against the profile's
     * {@code baseUrl} and nothing else.
     *
     * <p>The CB-592 shadow is put in <em>last</em>, after the profile's own {@code env:}, so no
     * profile — present or future — can restore the admin token by naming it in config. This is
     * the one place the shadow is applied: every {@code buildLaunch} in every adapter calls this
     * first, so a new profile, and a peer kind not yet written, gets it for free.
     */
    protected Map<String, String> baseEnv(BridgedConfig.Profile cfg) {
        Map<String, String> workerEnv = new LinkedHashMap<>();
        String path = env.apply("PATH");
        if (path != null && !path.isBlank()) {
            workerEnv.put("PATH", path);
        }
        if (cfg != null && cfg.env() != null) {
            workerEnv.putAll(cfg.env());
        }
        workerEnv.put("GITEA_ACCESS_TOKEN", BLOCKED_GITEA_ACCESS_TOKEN);
        return workerEnv;
    }

    /** Defensive copy of {@code argv} plus room to append launch flags. */
    protected static List<String> mutableArgv(List<String> argv) {
        return new ArrayList<>(argv);
    }

    /**
     * Uninterruptible sleep — the production {@link #sleeper}. Tests supply their own no-op /
     * fast-faking sleeper so they never real-sleep.
     */
    protected static void sleepUninterruptibly(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            // preserve the interrupt flag but continue — poll loops should not be aborted by an
            // interrupt that was not meant for them.
        }
    }
}