fleetd/bridged/src/main/java/dev/ltms/bridged/lead/LeadLauncher.java

package dev.ltms.bridged.lead;

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 org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * Starts the leads {@code fleet.leaders:} declares, when none is already running (CB-558).
 *
 * <p><strong>A lead is not a member, and this class exists to keep it that way.</strong> Every other
 * spawn path in the daemon goes through {@code HerdrPeerLauncher}, which does three things a lead
 * must never receive:
 * <ol>
 *   <li>it appends the <em>reply charter</em> — "you are an off-subscription worker … end every turn
 *       with {@code bridge_reply}". A lead is the orchestrator; telling it that it is a worker is
 *       exactly backwards.</li>
 *   <li>it registers the session with {@code SessionManager}, which subjects it to the idle reaper,
 *       the context cap and the shutdown drain. An idle lead is the normal state of a lead, so the
 *       reaper would kill the orchestrator for doing its job.</li>
 *   <li>it can move a peer off the subscription via {@code ANTHROPIC_BASE_URL}. A lead stays on the
 *       operator's subscription, always.</li>
 * </ol>
 * So this launcher talks to {@link AgentControl}/{@link WorkspaceControl} directly. The duplication
 * with the member launchers (argv and env assembly) is deliberate and is the cheaper half of the
 * trade: entangling the worker path with a not-a-worker case is how the three rules above get
 * broken later, quietly.
 *
 * <p><strong>Liveness, and the label round-trip.</strong> {@code LeadTabScanner} used to be able to
 * promise that bridged never writes a lead label. That is no longer true — an auto-launched lead is
 * labelled by this class, and the scanner reads that label back. The risk this opens is not
 * privilege escalation (the tab label never granted anything a pane could take for itself; see that
 * class's javadoc), but <em>staleness</em>: a label left behind by a crashed session would otherwise
 * read as a live lead forever, and the lead would never be relaunched. So a lead counts as live only
 * when herdr also reports a <em>running agent</em> in that tab — see {@link #liveLeads}. A labelled
 * tab with no agent in it is not a lead.
 */
public final class LeadLauncher {

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

    private final AgentControl agents;
    private final WorkspaceControl spaces;
    private final BridgedConfig cfg;

    /**
     * @param agents herdr agent control (start, list)
     * @param spaces workspace / tab control (ensure, create, label, list)
     * @param cfg    the loaded config — {@code fleet.leaders}, {@code profiles} and the lead pins
     */
    public LeadLauncher(AgentControl agents, WorkspaceControl spaces, BridgedConfig cfg) {
        this.agents = agents;
        this.spaces = spaces;
        this.cfg = cfg;
    }

    /**
     * Bring every declared lead up to its {@code instances} count, and return how many were started.
     *
     * <p>Never throws: a daemon that cannot start a lead must still serve. herdr being unreachable,
     * a profile that does not exist, a failed {@code agent.start} — each is logged and skipped, and
     * the remaining leads are still attempted.
     */
    public int ensureLeads() {
        Map<String, BridgedConfig.Leader> leaders = cfg.fleet().leaders();
        if (leaders.isEmpty()) {
            return 0;
        }

        Map<String, Integer> live;
        try {
            live = liveLeads(leaders);
        } catch (HerdrException e) {
            // Counting is the whole safety mechanism against double-spawning. If we cannot count, we
            // must not guess — spawning a second orchestrator is worse than starting none.
            log.warn("lead auto-launch skipped — cannot tell which leads are live: {}", e.getMessage());
            return 0;
        }

        int started = 0;
        for (Map.Entry<String, BridgedConfig.Leader> e : leaders.entrySet()) {
            String name = e.getKey();
            BridgedConfig.Leader lead = e.getValue();
            int running = live.getOrDefault(name, 0);
            int wanted = lead.instances();

            if (running >= wanted) {
                log.info("lead '{}': {} live, {} wanted — nothing to start", name, running, wanted);
                continue;
            }
            if (!lead.isCreatable()) {
                // A lead with a `terminal:` pin and no `profile:` is recognise-only by design: the
                // operator opens it by hand. Say so once rather than looking like a silent failure.
                log.info("lead '{}' is not live, and names no profile — it can be recognised but not "
                        + "launched. Add `profile:` under fleet.leaders.{} to have bridged start it.",
                        name, name);
                continue;
            }

            BridgedConfig.Profile profile = cfg.profiles().get(lead.profile());
            if (profile == null) {
                log.warn("lead '{}' names profile '{}', which is not configured — not launching",
                        name, lead.profile());
                continue;
            }

            for (int i = running; i < wanted; i++) {
                if (launch(name, lead, profile)) {
                    started++;
                }
            }
        }
        return started;
    }

    /**
     * How many live leads exist per configured name.
     *
     * <p>Two independent pieces of evidence, because either alone double-spawns:
     * <ul>
     *   <li>a running agent in a tab labelled {@code "<tabPrefix> <name>"} — how an auto-launched
     *       lead, or an operator following the labelling convention, is found;</li>
     *   <li>a running agent on a terminal the config pins in {@code fleet.leaders.<name>.terminal} —
     *       how a lead the operator opened and pinned by hand is found. Without this, a pinned lead
     *       whose tab carries no matching label would be relaunched on every boot.</li>
     * </ul>
     * Member workspaces are excluded, exactly as the scanner excludes them: a member must not be
     * counted as a lead because it happens to sit in a matching tab.
     */
    private Map<String, Integer> liveLeads(Map<String, BridgedConfig.Leader> leaders) {
        Set<String> memberSpaces = cfg.profiles().values().stream()
                .map(BridgedConfig.Profile::workspace)
                .filter(w -> w != null && !w.isBlank())
                .collect(Collectors.toSet());

        // tabId → the lead name its label declares.
        Map<String, String> nameByTab = new LinkedHashMap<>();
        for (Workspace ws : spaces.listWorkspaces()) {
            if (ws.workspaceId() == null || memberSpaces.contains(ws.label())) {
                continue;
            }
            for (Tab tab : spaces.listTabs(ws.workspaceId())) {
                String declared = leadNameOf(tab.label(), leaders);
                if (declared != null && tab.tabId() != null) {
                    nameByTab.put(tab.tabId(), declared);
                }
            }
        }

        // terminalId → the lead name the config pins it to.
        Map<String, String> nameByPinnedTerminal = new LinkedHashMap<>();
        leaders.forEach((name, lead) -> {
            if (lead.terminal() != null && !lead.terminal().isBlank()) {
                nameByPinnedTerminal.put(lead.terminal().strip(), name);
            }
        });

        Map<String, Integer> counts = new LinkedHashMap<>();
        for (Agent a : agents.list()) {
            String name = nameByTab.get(a.tabId());
            if (name == null) {
                name = nameByPinnedTerminal.get(a.terminalId());
            }
            if (name != null) {
                counts.merge(name, 1, Integer::sum);
            }
        }
        return counts;
    }

    /**
     * The configured lead a tab label names, or {@code null} for a label that names none.
     *
     * <p>Matched against the declared lead names rather than by splitting on the prefix, so an
     * operator's {@code "lead: something-else"} tab is not mistaken for a configured lead.
     */
    private String leadNameOf(String label, Map<String, BridgedConfig.Leader> leaders) {
        if (label == null) {
            return null;
        }
        String l = label.strip();
        for (Map.Entry<String, BridgedConfig.Leader> e : leaders.entrySet()) {
            if (l.equalsIgnoreCase(e.getValue().tabLabel(e.getKey()).strip())) {
                return e.getKey();
            }
        }
        return null;
    }

    /** Start one lead. Returns false (having logged) rather than throwing on any failure. */
    private boolean launch(String name, BridgedConfig.Leader lead, BridgedConfig.Profile profile) {
        String label = lead.tabLabel(name);
        String cwd = (lead.cwd() == null || lead.cwd().isBlank())
                ? System.getProperty("user.dir") : lead.cwd();

        Tab.Created tab = null;
        try {
            Workspace ws = spaces.ensureWorkspace(lead.workspace());
            tab = spaces.createTab(ws.workspaceId(), cwd, leadEnv(profile));
            if (tab.rootPaneId() == null) {
                throw new IllegalStateException("tab " + tab.tab().tabId()
                        + " came back with no seed pane — nowhere to start the lead");
            }
            // Same shape as the member launchers: herdr resolves the executable from `kind`, so
            // argv[0] (the configured launcher, e.g. `ccs`) is dropped and only the rest is passed.
            List<String> argv = leadArgv(profile);
            Agent started = agents.start("lead-" + name, herdrKind(profile),
                    argv.isEmpty() ? argv : argv.subList(1, argv.size()), tab.rootPaneId());

            // Label AFTER the start succeeds. A label written before would survive a failed start
            // and then read back as a live lead on the next boot, which is the exact staleness the
            // agent-liveness check exists to prevent — no need to create the case ourselves.
            spaces.renameTab(tab.tab().tabId(), label);

            log.info("lead '{}' launched: profile={} tab={} pane={} terminal={} label='{}' cwd={}",
                    name, profile.profile(), tab.tab().tabId(), started.paneId(),
                    started.terminalId(), label, cwd);
            return true;
        } catch (RuntimeException e) {
            log.warn("lead '{}' failed to launch on profile '{}': {}",
                    name, profile.profile(), e.getMessage());
            if (tab != null && tab.tab() != null && tab.tab().tabId() != null) {
                try {
                    spaces.closeTab(tab.tab().tabId());
                } catch (RuntimeException cleanup) {
                    log.warn("could not close the orphaned lead tab {}: {}",
                            tab.tab().tabId(), cleanup.getMessage());
                }
            }
            return false;
        }
    }

    /**
     * The herdr agent kind for this profile — the same value the matching member adapter passes, so
     * herdr resolves the same executable for a lead as it does for a member on that backend.
     */
    private static String herdrKind(BridgedConfig.Profile profile) {
        return profile.isOpenCode() ? "opencode" : "claude";
    }

    /**
     * The lead's argv: the profile's own command, the model pin, and the bridge MCP mount.
     *
     * <p>No {@code --append-system-prompt}. That flag carries the worker reply charter, and a lead
     * is not a worker — it reads its orchestration rules from the project's {@code CLAUDE.md} like
     * any other primary. This is the single most important difference from the member launchers;
     * do not "unify" it back.
     */
    private List<String> leadArgv(BridgedConfig.Profile profile) {
        List<String> argv = new ArrayList<>(profile.argv());
        if (profile.hasMcp()) {
            argv.add("--mcp-config");
            argv.add("{\"mcpServers\":{\"bridge\":{\"type\":\"http\",\"url\":\""
                    + profile.mcpUrl() + "\"}}}");
        }
        // Appended last, for the same reason the member launcher does it (CB-533): the argv is
        // usually a wrapper such as `ccs <profile>`, which exports its own model family over
        // whatever it inherited, and --model outranks the environment.
        if (profile.model() != null && !profile.model().isBlank()) {
            argv.add("--model");
            argv.add(profile.model());
        }
        return argv;
    }

    /**
     * The lead's environment: the profile's {@code env:} block, and nothing that could move it off
     * the subscription.
     *
     * <p>{@code ANTHROPIC_BASE_URL} and {@code ANTHROPIC_AUTH_TOKEN} are stripped unconditionally —
     * not defaulted, not guarded, stripped. A lead runs on the operator's subscription by
     * definition, so there is no configuration under which pointing it elsewhere is correct, and a
     * profile that carries them (a member profile reused as a lead's backend) must not leak them in.
     */
    private Map<String, String> leadEnv(BridgedConfig.Profile profile) {
        Map<String, String> out = new LinkedHashMap<>();
        if (profile.env() != null) {
            out.putAll(profile.env());
        }
        out.remove("ANTHROPIC_BASE_URL");
        out.remove("ANTHROPIC_AUTH_TOKEN");
        if (profile.configDir() != null && !profile.configDir().isBlank()) {
            out.put("CLAUDE_CONFIG_DIR", profile.configDir());
        }
        // Deliberately no git token: a lead reviews and merges through the operator's own
        // credentials, and never needs the scoped write:repository token a member is granted.
        out.values().removeIf(Objects::isNull);
        return out;
    }
}