Party Bus

2026-10-03  pace EASY  energy 71
   -> PHOTO_WALK by photo-walk-planner: Golden-hour walk with a 4-shot list
2026-10-04  pace FULL  energy 49
   !! WEATHER: Atlantic squall parks over the city until mid-afternoon.
   -> DINING by dining-planner: Market morning, long lunch, dinner at Taberna da Rua das Flores
2026-10-05  pace EASY  energy 33
   !! ILLNESS: Marco ordered the clams. Marco regrets the clams.
   -> SIGHTS by sights-planner: Sights day: Sintra and Pena Palace and the streets around it
2026-10-06  pace REST  energy 29
   !! PROTEST: Transport unions march on the Assembleia; Baixa is sealed off.
   -> REST by rest-planner: Rest day: pool, siesta, one gentle dinner
2026-10-07  pace REST  energy 23
   !! STRIKE: Metro and ferry strike; the tram to Belém is a rumor.
   -> REST by rest-planner: Rest day: pool, siesta, one gentle dinner
2026-10-08  pace REST  energy 25
   -> REST by rest-planner: Rest day: pool, siesta, one gentle dinner
2026-10-09  pace REST  energy 15
   !! LOST_PASSPORT: Priya's passport is in a taxi. Somewhere.
   -> REST by rest-planner: Rest day: pool, siesta, one gentle dinner

== the week, as the space remembers it ==
advisory [baseline] GO  GO=3 CAUTION=0 AVOID=0
advisory [lisbon-2026-d3-protest] CAUTION  GO=0 CAUTION=2 AVOID=1
booked lisbon-2026-stay-2 PB-643B6983 by clerk-b at log index 3
booked lisbon-2026-stay-1 PB-FC5A3CA3 by clerk-a at log index 1
next time: Porto and the Douro  [Porto and the Douro=2, Seville=1, The Azores=0]
plan: "Seven Hills, Three Friends, One Pool"  advisory GO  2 stays  https://www.google.com/maps/dir/Lisbon%2C%2BPortugal/Sintra%2C%2BPortugal

public record TravelBrief(String tripId, String from, String to, String departureDate,
                          String returnDate, double dailyBudget, String brief,
                          String transportPreference) {

    /** The trip's days as ISO-8601 strings, departure to return inclusive. */
    public List<String> days() {
        LocalDate start = LocalDate.parse(departureDate);
        LocalDate end = LocalDate.parse(returnDate);
        return start.datesUntil(end.plusDays(1)).map(LocalDate::toString).toList();
    }

    /** The brief as prompt text, the way Tripper's {@code contribution()} renders it. */
    public String contribution() {
        return "Journey from " + from + " to " + to + "\nDates: " + departureDate + " to "
                + returnDate + "\nBrief: " + brief + "\nTransport preference: "
                + transportPreference + "\nDaily budget (USD): " + dailyBudget;
    }
}

public record Traveler(String name, String about, List<String> needs) {
}

public record FlightRequest(String tripId, String topic, String from, String to,
                            String departureDate, String returnDate, String preference) {
}

public record FlightOptions(String tripId, String topic, List<FlightOption> options,
                            String note, String author) {
}

/**
 * Starts one peer: joins the group over TCP, builds every space, and wires
 * the capabilities.
 *
 * @param name     the peer's name (its agents' host)
 * @param port     the TCP port to listen on
 * @param seedPort an existing member's port, or 0 for the first member
 * @return the running peer
 * @throws Exception if the port cannot be bound
 */
public static Peer startPeer(String name, int port, int seedPort) throws Exception {
    Objects.requireNonNull(name, "name");
    PeerIdentity identity = PeerIdentity.generate();
    // Channel authentication (SPEC §5.6): TLS 1.3 with identity-endorsed
    // certificates, so the channel vouches for the sender and frames travel
    // unsigned. Fifteen peers times nine spaces is thousands of gossip frames
    // a second; signing each one would spend the machine on Ed25519.
    PeerNode node = PeerNode.builder(identity)
            .channelAuth(PeerNode.ChannelAuth.ATTESTED).build();
    node.listen(new TlsTcpTransport(identity), HOST + ":" + port);
    List<PeerAdvertisement.Endpoint> seeds = seedPort == 0 ? List.of()
            : List.of(new PeerAdvertisement.Endpoint("tls", HOST + ":" + seedPort, 0));
    GroupRuntime runtime = node.joinGroup(group(),
            new GroupMembership.Config(Duration.ofSeconds(30), Duration.ofSeconds(2), 2),
            seeds);
    CborCodec codec = CborCodec.defaultCodec();
    InstantSource clock = InstantSource.system();
    DiscoveryService discovery = new DiscoveryService(runtime, new AdCache(codec, clock),
            codec, identity.peerId());

    Map<String, ReplicatedSpace> spaces = new LinkedHashMap<>();
    for (String spaceName : SPACES) {
        ReplicatedSpace.Builder builder = ReplicatedSpace.builder(runtime, spaceName,
                identity, name).settleWindow(Duration.ofMillis(200));
        if (spaceName.equals(SLOTS)) {
            // The bid comes from the planner's @BidFunction at bind time; a bid
            // function is single-assignment per space per peer (QA4 A4-8), so a
            // placeholder here would be the first bidder and the planner refused.
            builder.strategy(ConflictStrategyType.AUCTION);
        }
        spaces.put(spaceName, builder.build());
    }

    AgentBinder binder = new AgentBinder(identity, runtime.id(), discovery, clock);
    spaces.forEach(binder::space);

    CapabilityPipes pipes = new CapabilityPipes(runtime, codec);
    PushSumAggregate aggregate = new PushSumAggregate(pipes, runtime.sampler(),
            identity.peerId(), codec, clock);
    VoteCapability advisoryVote = new VoteCapability(spaces.get(ADVISORIES),
            identity.agent(name), identity.peerId(), clock);
    VoteCapability partyVote = new VoteCapability(spaces.get(DECISIONS),
            identity.agent(name), identity.peerId(), clock);
    // What the annotations cast and watch with: @Ballot/@OnDecision on the two
    // vote spaces, and a Contribution return on the aggregate.
    binder.vote(ADVISORIES, advisoryVote).vote(DECISIONS, partyVote).aggregate(aggregate);

    // One clock: the runtime rides the peer tick, so registering is all it
    // takes for the aggregate to gossip and for every capability to be found.
    CapabilityRuntime capabilities = new CapabilityRuntime(runtime, discovery, identity);
    capabilities.register(aggregate);
    capabilities.register(advisoryVote);
    capabilities.register(partyVote);
    node.startTicking(CLERK_TICK);
    return new Peer(name, identity, node, runtime, discovery, spaces, binder, codec, pipes,
            capabilities, aggregate, advisoryVote, partyVote, port);
}

<dependency>
      <groupId>ai.badmonkey.agentspaces</groupId>
      <artifactId>agentspaces-capabilities</artifactId>
      <version>${agentspaces.version}</version>
    </dependency>
    <dependency>
      <groupId>ai.badmonkey.agentspaces</groupId>
      <artifactId>agentspaces-agent</artifactId>
      <version>${agentspaces.version}</version>
    </dependency>
    <dependency>
      <groupId>ai.badmonkey.agentspaces</groupId>
      <artifactId>agentspaces-console</artifactId>
      <version>${agentspaces.version}</version>
    </dependency>
    <!--
      The Embabel extension reads Embabel's annotations reflectively and needs
      no Embabel artifact to compile. The planner bridge it carries generates
      the fleet agent the real Embabel planner plans over (see the embabel
      profile below and src/embabel/java).
    -->
    <dependency>
      <groupId>ai.badmonkey.agentspaces</groupId>
      <artifactId>embabel-agentspaces</artifactId>
      <version>${agentspaces.version}</version>
    </dependency>

public static final class FlightScout {
    private final Sources sources;

    public FlightScout(Sources sources) {
        this.sources = Objects.requireNonNull(sources, "sources");
    }

    @SpaceTake(space = SPACE, lease = "2m", pollTimeout = "500ms")
    public FlightOptions scout(FlightRequest request) {
        List<Link> web = sources.search("flights " + request.from() + " to " + request.to()
                + " " + request.departureDate() + " return " + request.returnDate(), 8);
        FlightOptions found = sources.structured("flights", RESEARCHER + "\n"
                + "Propose up to three itineraries from " + request.from() + " to "
                + request.to() + " departing " + request.departureDate() + " returning "
                + request.returnDate() + ". Preference: " + request.preference() + ".\n"
                + digest("Search results", web), FlightOptions.class);
        return new FlightOptions(request.tripId(), request.topic(), found.options(),
                found.note(), "flight-scout");
    }
}

public static final class HospitalLocator {
    private final Sources sources;

    public HospitalLocator(Sources sources) {
        this.sources = Objects.requireNonNull(sources, "sources");
    }

    @SpaceTake(space = Scouts.SPACE, lease = "1m", pollTimeout = "500ms")
    public HospitalList locate(HospitalRequest request) {
        List<Sources.Place> places = sources.places(request.location(), "hospital", 5);
        List<Hospital> hospitals = new ArrayList<>();
        for (Sources.Place place : places) {
            hospitals.add(new Hospital(place.name(), place.address(),
                    Math.round(distanceFromFirst(places, place) * 10.0) / 10.0,
                    place.emergency(), place.phone()));
        }
        return new HospitalList(request.tripId(), request.topic(), request.location(),
                hospitals, emergencyNumber(request.location()), "hospital-locator");
    }

    private static double distanceFromFirst(List<Sources.Place> places, Sources.Place place) {
        if (places.isEmpty()) {
            return 0.0;
        }
        Sources.Place origin = places.get(0);
        return ai.badmonkey.agentspaces.partybus.sources.LiveSources.distanceKm(
                origin.lat(), origin.lon(), place.lat(), place.lon());
    }

    /** The emergency number: 112 across Europe, 911 in North America, else 112. */
    static String emergencyNumber(String location) {
        String lower = location.toLowerCase();
        if (lower.contains("usa") || lower.contains("united states") || lower.contains(", us")
                || lower.contains("canada") || lower.contains("mexico")) {
            return "911";
        }
        return "112";
    }
}

public interface Sources {

    /** A place from the map provider. */
    record Place(String name, String address, double lat, double lon, String phone,
                 boolean emergency) {
    }

    /** A short label for logs and the console: {@code live} or {@code recorded}. */
    String name();

    /** Web search results. */
    List<Link> search(String query, int limit);

    /** News results, freshest first. */
    List<Link> news(String query, int limit);

    /** Image results; each link is the image itself. */
    List<Link> images(String query, int limit);

    /** Video results, YouTube first when a key is present. */
    List<Link> videos(String query, int limit);

    /**
     * Places of one kind near a location, from the map provider.
     *
     * @param location a place name to geocode
     * @param amenity  an OpenStreetMap amenity value, e.g. {@code hospital}
     * @param limit    the maximum results
     * @return the places, nearest first
     */
    List<Place> places(String location, String amenity, int limit);

    /**
     * Asks the language model for a structured answer and binds it to a record
     * type. The prompt describes the task; the implementation adds the shape of
     * the record and parses the JSON that comes back.
     *
     * @param tag    a stable name for the kind of answer (the recorded key)
     * @param prompt the task
     * @param type   the record type to bind
     * @param <T>    the record type
     * @return the bound answer
     */
    <T> T structured(String tag, String prompt, Class<T> type);

    /**
     * The sources the environment selects: {@link LiveSources} when the API keys
     * are present, and otherwise a fail-fast error naming what is missing, the
     * way Tripper insists on its keys. {@code PARTYBUS_SOURCES=recorded} selects
     * the recorded trip explicitly.
     *
     * @return the configured sources
     */
    static Sources fromEnvironment() {
        String mode = System.getenv().getOrDefault("PARTYBUS_SOURCES", "live");
        if ("recorded".equalsIgnoreCase(mode)) {
            return RecordedSources.lisbon();
        }
        return LiveSources.fromEnvironment();
    }
}

public static LiveSources fromEnvironment(Map<String, String> env) {
    List<String> missing = new ArrayList<>();
    for (String required : List.of(OPENAI_API_KEY, BRAVE_API_KEY)) {
        if (env.getOrDefault(required, "").isBlank()) {
            missing.add(required);
        }
    }
    if (!missing.isEmpty()) {
        throw new MissingApiKeyException(missing);
    }
    OpenAiChatModel model = OpenAiChatModel.builder()
            .options(OpenAiChatOptions.builder()
                    .apiKey(env.get(OPENAI_API_KEY))
                    .model(env.getOrDefault(OPENAI_MODEL, "gpt-4.1-mini"))
                    .temperature(0.2)
                    .build())
            .build();
    return new LiveSources(
            HttpClient.newBuilder().connectTimeout(Duration.ofSeconds(15)).build(),
            ChatClient.create(model), env.get(BRAVE_API_KEY),
            Optional.ofNullable(env.get(YOUTUBE_API_KEY)).filter(k -> !k.isBlank()));
}

public <T> T structured(String tag, String prompt, Class<T> type) {
    // Spring AI's structured output: the response format comes from the
    // target type, and the answer binds straight into the record.
    T answer = chat.prompt()
            .system("You are a specialist on a travel-planning team. Use real, current"
                    + " information from the material you are given; never invent URLs.")
            .user(prompt)
            .call()
            .entity(type);
    if (answer == null) {
        throw new IllegalStateException("the model returned no " + type.getSimpleName() + " for " + tag);
    }
    return answer;
}

public <T> T structured(String tag, String prompt, Class<T> type) {
    count(tag);
    int colon = tag.indexOf(':');
    String base = colon < 0 ? tag : tag.substring(0, colon);
    JsonNode node = lookup(recording.path("structured"), tag, base);
    if (node.isMissingNode()) {
        throw new IllegalStateException("recording has no structured answer for '" + tag
                + "' (nor '" + base + "')");
    }
    return Json.bind(node, type);
}

public void dispatch(TravelBrief brief) {
    String id = brief.tripId();
    Travelers party = trip.read(Template.of(Travelers.class).where("tripId", eq(id)),
            Duration.ofSeconds(10)).orElseThrow(() -> new IllegalStateException(
            "no travelers written for trip " + id));
    List<String> needs = party.allNeeds();
    String who = party.contribution();
    double nightly = brief.dailyBudget() / 2.0; // Tripper's half-the-budget rule

    research.write(new FlightRequest(id, "flights " + brief.from() + " to " + brief.to(),
            brief.from(), brief.to(), brief.departureDate(), brief.returnDate(),
            brief.transportPreference()), REQUEST_LEASE);
    research.write(new HotelRequest(id, "hotels " + brief.to(), brief.to(),
            brief.departureDate(), brief.returnDate(), nightly, needs), REQUEST_LEASE);
    research.write(new CarRequest(id, "car " + brief.to(), brief.to(), brief.departureDate(),
            brief.returnDate(), party.travelers().size(), needs), REQUEST_LEASE);
    research.write(new SightseeingRequest(id, "sights " + brief.to(), brief.to(),
            brief.brief(), brief.departureDate(), brief.returnDate(), who), REQUEST_LEASE);
    research.write(new RestaurantRequest(id, "restaurants " + brief.to(), brief.to(),
            brief.dailyBudget() / 4.0, needs), REQUEST_LEASE);
    research.write(new DietRequest(id, "diet " + brief.to(), brief.to(), needs),
            REQUEST_LEASE);
    research.write(new FoodieRequest(id, "foodie " + brief.to(), brief.to(), brief.brief()),
            REQUEST_LEASE);
    research.write(new PhotoRequest(id, "photos " + brief.to(), "landmarks", brief.to()),
            REQUEST_LEASE);
    research.write(new ReviewRequest(id, "reviews " + brief.to(), "visiting", brief.to()),
            REQUEST_LEASE);
    research.write(new AccessibilityRequest(id, "accessibility " + brief.to(), brief.to(),
            needs), REQUEST_LEASE);
    research.write(new HospitalRequest(id, "hospitals " + brief.to(), brief.to(),
            "pre-trip: know the nearest emergency room"), REQUEST_LEASE);
    research.write(new NextDestinationRequest(id, "next after " + brief.to(), brief.to(),
            brief.brief(), who), REQUEST_LEASE);
    sweep(id, brief.to(), "travel safety", Safety.BASELINE);

    List<String> interests = List.of(brief.brief().split("[,;.]\\s*"));
    int day = 0;
    for (String date : brief.days()) {
        slots.write(new DaySlot(id, date, brief.to(), interests, day == 0 ? "EASY" : "FULL", 0),
                REQUEST_LEASE);
        day++;
    }
}

public EventResponse react(TripEvent event) {
    String id = event.tripId();
    switch (event.kind()) {
        case ILLNESS -> {
            research.write(new HospitalRequest(id, "hospitals " + event.eventId(),
                    event.location(), event.detail()), REQUEST_LEASE);
            research.write(new DietRequest(id, "diet " + event.eventId(), event.location(),
                    List.of("recovering from food poisoning", "bland food")), REQUEST_LEASE);
            return response(event, "asked the hospital locator for the nearest ER and the "
                    + "diet advisor for a gentle menu");
        }
        case PROTEST -> {
            sweep(id, event.location(), "protest unrest", event.eventId());
            return response(event, "opened a fresh safety sweep on all three channels; "
                    + "the panel re-votes the advisory");
        }
        case STRIKE -> {
            reauction(id, event.date());
            research.write(new CarRequest(id, "car " + event.eventId(), event.location(),
                    event.date(), event.date(), 3, List.of()), REQUEST_LEASE);
            return response(event, "re-auctioned the day and re-scouted ground transport");
        }
        case WEATHER -> {
            reauction(id, event.date());
            return response(event, "re-auctioned the day; outdoor bids rise, indoor bids win");
        }
        case LOST_PASSPORT -> {
            research.write(new AccessibilityRequest(id, "consulate " + event.eventId(),
                    event.location(), List.of("emergency travel document", "nearest consulate")),
                    REQUEST_LEASE);
            return response(event, "asked the accessibility advisor for the consulate route");
        }
        default -> {
            return response(event, "a calm day; nothing to do");
        }
    }
}

public static final class PlanAssembler {
    private final Sources sources;

    public PlanAssembler(Sources sources) {
        this.sources = Objects.requireNonNull(sources, "sources");
    }

    @SpaceJoin(space = Party.TRIP, key = "tripId", resultLease = "12h",
            parts = {@Part(TravelBrief.class),
                     @Part(value = SightseeingOptions.class, space = Scouts.SPACE),
                     @Part(value = HotelOptions.class, space = Scouts.SPACE),
                     @Part(value = RestaurantOptions.class, space = Scouts.SPACE),
                     @Part(value = FoodieOptions.class, space = Scouts.SPACE),
                     @Part(value = FlightOptions.class, space = Scouts.SPACE),
                     @Part(value = Travelers.class, optional = true)})
    public ProposedTravelPlan assemble(Joined j) {
        TravelBrief brief = j.get(TravelBrief.class);
        Travelers party = j.find(Travelers.class).orElse(new Travelers(j.key(), List.of()));
        ProposedTravelPlan drafted = sources.structured("plan", PLANNER + "\n"
                + "Write the plan for this brief as HTML starting at <h4>, in 500 "
                + "words or fewer, with a catchy title, one Day entry per date in "
                + "'City,+Country' form, and the countries visited.\n<brief>"
                + brief.contribution() + "</brief>\n" + party.contribution()
                + "\nPoints of interest: " + j.get(SightseeingOptions.class).pointsOfInterest()
                + "\nHotels: " + j.get(HotelOptions.class).options()
                + "\nRestaurants: " + j.get(RestaurantOptions.class).restaurants()
                + "\nMust eat: " + j.get(FoodieOptions.class).dishes()
                + "\nFlights: " + j.get(FlightOptions.class).options(), ProposedTravelPlan.class);
        List<Day> days = drafted.days().isEmpty() ? defaultDays(brief) : drafted.days();
        return new ProposedTravelPlan(j.key(), drafted.title(), drafted.plan(), days,
                drafted.countriesVisited(), "plan-assembler");
    }
}

public TravelPlan finish(ProposedTravelPlan plan) {
    Map<String, List<String>> byTown = new LinkedHashMap<>();
    for (Day day : plan.days()) {
        byTown.computeIfAbsent(day.stayingAt(), k -> new ArrayList<>()).add(day.date());
    }
    List<HotelOptions> hotels = research.readAll(Template.of(HotelOptions.class)
            .where("tripId", eq(plan.tripId())), 20);
    List<Stay> stays = new ArrayList<>();
    int order = 0;
    for (Map.Entry<String, List<String>> e : byTown.entrySet()) {
        Optional<HotelOption> pick = hotels.stream()
                .filter(h -> h.location().toLowerCase().startsWith(e.getKey().toLowerCase())
                        || e.getKey().toLowerCase().startsWith(h.location().toLowerCase()))
                .flatMap(h -> h.options().stream()).findFirst()
                .or(() -> hotels.stream().flatMap(h -> h.options().stream()).findFirst());
        Stay stay = new Stay(e.getKey(), e.getValue(),
                pick.map(HotelOption::name).orElse("lodging still being scouted"),
                pick.map(HotelOption::bookingUrl).orElse(""),
                pick.map(HotelOption::nightlyUsd).orElse(0.0));
        stays.add(stay);
        if (pick.isPresent()) {
            bookings.write(new BookingOrder(plan.tripId(),
                    plan.tripId() + "-stay-" + (++order), "HOTEL", stay.lodging(),
                    e.getValue().get(0), stay.nightly() * e.getValue().size()),
                    Lease.of(Duration.ofHours(12)));
        }
    }
    // The panel's baseline vote usually lands before the plan; give it a
    // moment when it has not, and ship PENDING rather than wait forever.
    String advisory = trip.read(Template.of(TravelAdvisory.class)
                    .where("tripId", eq(plan.tripId())), Duration.ofSeconds(45))
            .map(TravelAdvisory::verdict).orElse("PENDING");
    return new TravelPlan(plan.tripId(), plan.title(), plan.plan(), plan.days(), stays,
            advisory, mapUrl(plan.days()));
}

public static final class NewsWatcher {
    private final Sources sources;

    public NewsWatcher(Sources sources) {
        this.sources = Objects.requireNonNull(sources, "sources");
    }

    @SpaceTake(space = Scouts.SPACE, resultSpace = SIGNALS, lease = "1m",
            pollTimeout = "500ms")
    public SignalReport sweep(NewsSweep sweep) {
        List<Link> news = sources.news(sweep.location() + " " + sweep.topic(), 10);
        return new SignalReport(sweep.tripId(), sweep.topic(), sweep.location(), "news",
                sweep.wave(), news.stream().map(Link::summary).toList(), news,
                "news-watcher");
    }
}

public static final class SentimentAnalyst {
    private final Sources sources;

    public SentimentAnalyst(Sources sources) {
        this.sources = Objects.requireNonNull(sources, "sources");
    }

    @SpaceNotify(space = SIGNALS, lease = "12h")
    public DangerAssessment assess(SignalReport report) {
        DangerAssessment scored = sources.structured(
                "danger:" + report.channel() + ":" + report.topic(),
                "You assess danger to ordinary tourists, not to journalists or soldiers. "
                + "Score 0 (calm) to 100 (active conflict) and name the themes. Location: "
                + report.location() + ". Channel: " + report.channel() + ".\n"
                + digest("What the watcher found", report.sources()), DangerAssessment.class);
        return new DangerAssessment(report.tripId(), report.location(), report.channel(),
                report.wave(), scored.danger(), scored.themes(), "sentiment-analyst");
    }
}

public static double trust(String seat, String channel) {
    if (seat.equals(channel)) {
        return 1.0;
    }
    return switch (channel) {
        case "news" -> 0.9;
        case "social" -> 0.7;
        default -> 0.6; // video
    };
}

public static String verdictFor(double weightedDanger) {
    if (weightedDanger >= AVOID_AT) {
        return "AVOID";
    }
    return weightedDanger >= CAUTION_AT ? "CAUTION" : "GO";
}

public static final class Panelist {
    private final String seat;
    @SpaceRef(Watchers.SIGNALS)
    Space signals;

    public Panelist(String seat) {
        this.seat = Objects.requireNonNull(seat, "seat");
    }

    @Ballot(space = ADVISORIES, prefix = PROPOSAL_PREFIX, lease = "12h")
    public String judge(VoteCapability.Proposal proposal) {
        // The proposal names the wave; read its assessments, waiting briefly
        // for our own channel's when it is still in flight.
        long deadline = System.nanoTime() + Duration.ofSeconds(20).toNanos();
        while (System.nanoTime() < deadline) {
            List<DangerAssessment> wave = waveOf(proposal.proposalId());
            boolean ownSeen = wave.stream().anyMatch(a -> a.channel().equals(seat));
            if (ownSeen || wave.size() >= SEATS.size()) {
                double weighted = wave.stream()
                        .mapToDouble(a -> a.danger() * trust(seat, a.channel()))
                        .max().orElse(0.0);
                return verdictFor(weighted);
            }
            sleep(100);
        }
        return null; // the wave never arrived here: abstain
    }

    private List<DangerAssessment> waveOf(String proposalId) {
        return signals.readAll(Template.of(DangerAssessment.class), 500).stream()
                .filter(a -> Safety.proposalId(a.tripId(), a.location(), a.wave()).equals(proposalId))
                .toList();
    }
}

public static final class Lead {
    private final String leadName;
    private final Map<String, DangerAssessment> waves = new java.util.concurrent.ConcurrentHashMap<>();
    @SpaceRef(Watchers.SIGNALS)
    Space signals;

    public Lead(String leadName) {
        this.leadName = Objects.requireNonNull(leadName, "leadName");
    }

    /**
     * One vote per wave, as one annotation (ISSUE-Propose): the id is the
     * prefix plus {@code tripId:location:wave}, the format
     * {@link Safety#proposalId} has always used, so every channel's
     * assessment of one wave maps to one proposal and the binder invokes
     * this once per wave, opens once, and skips the rest.
     */
    @Propose(space = Watchers.SIGNALS, vote = ADVISORIES, prefix = PROPOSAL_PREFIX,
            key = {"tripId", "location", "wave"}, options = {"GO", "CAUTION", "AVOID"}, quorum = 3,
            lease = "12h", subscriptionLease = "12h")
    public String open(DangerAssessment assessment) {
        waves.putIfAbsent(Safety.proposalId(assessment.tripId(), assessment.location(),
                assessment.wave()), assessment);
        return "Travel advisory for " + assessment.location() + " (" + assessment.wave() + ")";
    }

    @OnDecision(space = ADVISORIES, prefix = PROPOSAL_PREFIX, resultSpace = Party.TRIP,
            resultLease = "12h", lease = "12h")
    public TravelAdvisory publish(VoteCapability.Decision decision) {
        DangerAssessment wave = waves.get(decision.proposalId());
        if (wave == null) {
            return null; // a vote opened elsewhere on a wave this lead never saw
        }
        Map<String, Integer> tally = decision.tally();
        Set<String> reasons = new TreeSet<>();
        signals.readAll(Template.of(DangerAssessment.class)
                .where("tripId", eq(wave.tripId()))
                .where("location", eq(wave.location()))
                .where("wave", eq(wave.wave())), 50)
                .forEach(a -> reasons.addAll(a.themes()));
        return new TravelAdvisory(wave.tripId(), wave.location(), wave.wave(), decision.winner(),
                tally.getOrDefault("GO", 0), tally.getOrDefault("CAUTION", 0),
                tally.getOrDefault("AVOID", 0), List.copyOf(reasons), leadName);
    }
}

public record DaySlot(String tripId, String date, String location, List<String> interests,
                      String pace, int round) {
}

public static final class DiningPlanner {
    @SpaceRef(Scouts.SPACE)
    Space research;

    @BidFunction(space = SLOTS)
    public double bid(DaySlot slot) {
        double bid = 90;
        if (interested(slot, "food", "wine", "eat", "cook")) {
            bid -= 35;
        }
        if (slot.pace().equals("REST")) {
            bid += 20;
        }
        return bid + slot.round() * 5; // a strike barely touches a lunch
    }

    @SpaceTake(space = SLOTS, lease = "1m", pollTimeout = "500ms")
    public ScheduledActivity plan(DaySlot slot) {
        String table = research.readAll(Template.of(RestaurantOptions.class)
                        .where("tripId", eq(slot.tripId())), 20).stream()
                .filter(r -> r.location().equalsIgnoreCase(slot.location()))
                .flatMap(r -> r.restaurants().stream()).map(r -> r.name()).findFirst()
                .orElse("a table the scouts are still finding");
        return new ScheduledActivity(slot.tripId(), slot.date(), slot.location(), "DINING",
                "Market morning, long lunch, dinner at " + table, bid(slot),
                "dining-planner", slot.round());
    }
}

public static final class SightsPlanner {
    @SpaceRef(Scouts.SPACE)
    Space research;

    @BidFunction(space = SLOTS)
    public double bid(DaySlot slot) {
        double bid = 80;
        if (interested(slot, "history", "architecture", "museum", "art")) {
            bid -= 35;
        }
        if (slot.pace().equals("REST")) {
            bid += 80; // nobody wants a six-mile walk on a rest day
        }
        return bid + slot.round() * 45; // storms and strikes ruin an outdoor day
    }

    @SpaceTake(space = SLOTS, lease = "1m", pollTimeout = "500ms")
    public ScheduledActivity plan(DaySlot slot) {
        List<PointOfInterest> pois = research.readAll(Template.of(SightseeingOptions.class)
                        .where("tripId", eq(slot.tripId())), 5).stream()
                .flatMap(s -> s.pointsOfInterest().stream()).toList();
        String pick = pois.isEmpty() ? "the old town"
                : pois.get(Math.abs(slot.date().hashCode()) % pois.size()).name();
        return new ScheduledActivity(slot.tripId(), slot.date(), slot.location(), "SIGHTS",
                "Sights day: " + pick + " and the streets around it", bid(slot),
                "sights-planner", slot.round());
    }
}

public static final class RestPlanner {
    @BidFunction(space = SLOTS)
    public double bid(DaySlot slot) {
        // Expensive to waste a good day; nearly free when the party is spent.
        return slot.pace().equals("REST") ? 5 : 140;
    }

    @SpaceTake(space = SLOTS, lease = "1m", pollTimeout = "500ms")
    public ScheduledActivity plan(DaySlot slot) {
        return new ScheduledActivity(slot.tripId(), slot.date(), slot.location(), "REST",
                "Rest day: pool, siesta, one gentle dinner", bid(slot), "rest-planner",
                slot.round());
    }
}

public static final class TravelerAgent {
    private final Traveler traveler;
    private final int stamina;
    private final PushSumAggregate aggregate;

    @SpaceRef(TRIP)
    Space trip;

    /**
     * @param traveler  who
     * @param stamina   baseline energy, 0 to 100
     * @param aggregate the push-sum capability this traveler reports into
     * @param vote      the vote capability over the decisions space
     */
    public TravelerAgent(Traveler traveler, int stamina, PushSumAggregate aggregate) {
        this.traveler = Objects.requireNonNull(traveler, "traveler");
        this.stamina = stamina;
        this.aggregate = Objects.requireNonNull(aggregate, "aggregate");
    }

    /** The traveler's name. */
    public String name() {
        return traveler.name();
    }

    /**
     * How this traveler feels on a day: stamina minus wear, minus a lot when
     * the day's events hit them, and the same number goes into the aggregate.
     */
    @SpaceNotify(space = TRIP, lease = "12h")
    public EnergyReport feel(TripDay day) {
        int energy = energyOn(day, trip.readAll(Template.of(TripEvent.class)
                .where("tripId", eq(day.tripId())).where("date", eq(day.date())), 20));
        aggregate.start(Itinerary.energyEpoch(day.tripId(), day.date()), energy);
        double spent = 40 + (traveler.about().toLowerCase().contains("food") ? 45 : 20)
                + day.dayIndex() * 3;
        aggregate.start(Itinerary.spendEpoch(day.tripId(), day.date()), spent);
        return new EnergyReport(day.tripId(), day.date(), traveler.name(), energy, spent);
    }

    /** Pure and deterministic, so the simulator's story replays the same way. */
    int energyOn(TripDay day, List<TripEvent> events) {
        int energy = stamina - day.dayIndex() * 9;
        for (TripEvent event : events) {
            if (event.kind() == EventKind.ILLNESS
                    && event.detail().toLowerCase().contains(traveler.name().toLowerCase())) {
                energy -= 60;
            } else if (event.kind() == EventKind.WEATHER || event.kind() == EventKind.STRIKE) {
                energy -= 10;
            } else if (event.kind() == EventKind.PROTEST) {
                energy -= 15;
            }
        }
        return Math.max(0, Math.min(100, energy));
    }

    /**
     * One signed ballot per proposal, for the option that sounds most like them.
     * The cue is the vote's own proposal, so the ballot is never refused as
     * unknown and no wait is needed.
     */
    @Ballot(space = DECISIONS, prefix = "next:", lease = "12h")
    public String decide(VoteCapability.Proposal proposal) {
        return preference(proposal.options());
    }

    String preference(List<String> options) {
        String about = traveler.about().toLowerCase();
        for (String option : options) {
            for (String word : option.toLowerCase().split("[^a-z]+")) {
                if (word.length() > 3 && about.contains(word)) {
                    return option;
                }
            }
        }
        // Fall back on a stable, name-derived choice so ties break differently per traveler.
        return options.get(Math.abs(traveler.name().hashCode()) % options.size());
    }
}

public static final class Lead {
    private final int partySize;
    @SpaceRef(DECISIONS)
    Space decisions;

    public Lead(int partySize) {
        this.partySize = partySize;
    }

    /**
     * The options come from the cue (the advisor's ideas), so this is the
     * dynamic form: the method returns the {@link Motion} and the binder opens
     * it once per proposal id (ISSUE-Motion). The human-readable announcement
     * is a second entry, written through the handle, since one return
     * carries one value.
     */
    @SpaceNotify(space = Scouts.SPACE, lease = "12h")
    public Motion open(NextDestinationIdeas ideas) {
        String proposalId = "next:" + ideas.tripId();
        List<String> options = new ArrayList<>();
        for (Destination idea : ideas.ideas()) {
            options.add(idea.name());
        }
        if (decisions.read(Template.of(NextDestinationProposal.class)
                .where("proposalId", eq(proposalId))).isEmpty()) {
            decisions.write(new NextDestinationProposal(ideas.tripId(), proposalId,
                    "Where next after this trip?", options, "party-lead"), Lease.of(Duration.ofHours(12)));
        }
        return Motion.in(DECISIONS, proposalId, "Where next after this trip?", options, partySize,
                Lease.of(Duration.ofHours(12)));
    }

    @OnDecision(space = DECISIONS, prefix = "next:", resultSpace = TRIP, resultLease = "12h",
            lease = "12h")
    public PartyDecision record(VoteCapability.Decision decision) {
        List<String> tally = new ArrayList<>();
        for (Map.Entry<String, Integer> e : decision.tally().entrySet()) {
            tally.add(e.getKey() + "=" + e.getValue());
        }
        return new PartyDecision(decision.proposalId().substring("next:".length()),
                decision.proposalId(), decision.winner(), tally);
    }
}

/**
 * Forms the booking quorum: a Raft group among the clerk peers, an
 * {@link OrderedTakes} coordinator per clerk over its {@code bookings}

public static final class BookingClerk {
    private final RaftLog raft;
    private final String clerk;

    public BookingClerk(RaftLog raft, String clerk) {
        this.raft = Objects.requireNonNull(raft, "raft");
        this.clerk = Objects.requireNonNull(clerk, "clerk");
    }

    @OrderedTake(space = BOOKINGS, lease = "30s", pollTimeout = "2s", resultLease = "12h")
    public BookingConfirmation confirm(BookingOrder order) {
        return new BookingConfirmation(order.tripId(), order.orderId(),
                confirmationCode(order, clerk), order.amountUsd(), clerk, raft.commitIndex());
    }
}

/**
 * The registry of remote actions over a peer's discovery and spaces, with
 * every request type routed to the research bag and every result awaited
 * there (sweeps excepted: watchers write their reports into {@code signals}).

/**
 * The planner's local repertoire: what a lone Tripper-style agent can do
 * with a brief before any scout is involved. Deliberately unable to reach a
 * {@code ReviewDigest} on its own.

public static Optional<List<Step>> plan(Class<?> start, Class<?> goal, List<Step> actions) {
    record Node(Class<?> type, List<Step> path) {
    }
    Deque<Node> frontier = new ArrayDeque<>();
    frontier.add(new Node(start, List.of()));
    Set<Class<?>> visited = new HashSet<>();
    visited.add(start);
    while (!frontier.isEmpty()) {
        Node node = frontier.poll();
        if (goal.isAssignableFrom(node.type())) {
            return Optional.of(node.path());
        }
        for (Step step : actions) {
            if (step.in().isAssignableFrom(node.type()) && visited.add(step.out())) {
                List<Step> path = new ArrayList<>(node.path());
                path.add(step);
                frontier.add(new Node(step.out(), path));
            }
        }
    }
    return Optional.empty();
}

planner alone reaches ReviewDigest: false
no Embabel on the classpath (-Pembabel adds it): planning over 17 remote actions directly
planner with the fleet: [askForSights, sightseeing-scout_SightseeingRequest,
                         askForReviewsOfTheTopSight, review-reader_ReviewRequest]
  -> reviews of the top sight: 4.6/5, praise: [the light on the river at 6pm,
     the food, and the price of the food, how kind people are when you attempt Portuguese]

// 6. The planner: unreachable alone, reachable and executed through the fleet.
            // The bridge snapshots the fleet's cards when it is built, so wait for the
            // two actions this plan needs, not merely for a dozen cards: with the front
            // half of the trip now settling in seconds, "twelve of thirty" no longer
            // implies the sightseeing scout's and review reader's cards are among them.
            RemoteActions remote = PartyBusPlanner.remoteActions(bus.plannerPeer());
            assertThat(PartyBusPlanner.awaitActions(remote, 12, Duration.ofSeconds(30)))
                    .isGreaterThanOrEqualTo(12);
            List<String> needed = List.of("sightseeing-scout_SightseeingRequest",
                    "review-reader_ReviewRequest");
            assertThat(await(() -> {
                Set<String> names = remote.available().stream().map(RemoteAction::name)
                        .collect(Collectors.toSet());
                return names.containsAll(needed) ? Optional.of(names) : Optional.<Set<String>>empty();
            }, Duration.ofSeconds(30)))
                    .as("the scout and reader actions the plan needs are advertised; seen: "
                            + remote.available().stream().map(RemoteAction::name).sorted().toList())
                    .isPresent();
            List<Step> local = PartyBusPlanner.localSteps(PartyBus.party(TRIP));
            assertThat(PartyBusPlanner.plan(TravelBrief.class, ReviewDigest.class, local)).isEmpty();
            EmbabelRemoteActions bridge = PartyBusPlanner.bridge(remote);
            List<Step> withFleet = new ArrayList<>(local);
            withFleet.addAll(PartyBusPlanner.fleetSteps(bridge.agentInstance().orElseThrow()));
            List<Step> found = PartyBusPlanner.plan(TravelBrief.class, ReviewDigest.class, withFleet)
                    .orElseThrow();
            assertThat(found).extracting(Step::name).containsExactly("askForSights",
                    "sightseeing_scout_SightseeingRequest", "askForReviewsOfTheTopSight",
                    "review_reader_ReviewRequest");
            ReviewDigest digest = (ReviewDigest) PartyBusPlanner.execute(found,
                    new TravelBrief(TRIP + "-plan", brief.from(), brief.to(), brief.departureDate(),
                            brief.returnDate(), brief.dailyBudget(), brief.brief(),
                            brief.transportPreference()));
            assertThat(digest.author()).isEqualTo("review-reader");
            assertThat(digest.averageRating()).isEqualTo(4.6);

package ai.badmonkey.agentspaces.partybus.embabel;

import ai.badmonkey.agentspaces.agent.remote.RemoteActions;
import ai.badmonkey.agentspaces.embabel.EmbabelRemoteActions;
import ai.badmonkey.agentspaces.partybus.PartyBus;
import ai.badmonkey.agentspaces.partybus.domain.Research.ReviewDigest;
import ai.badmonkey.agentspaces.partybus.domain.Research.ReviewRequest;
import ai.badmonkey.agentspaces.partybus.domain.Research.SightseeingOptions;
import ai.badmonkey.agentspaces.partybus.domain.Research.SightseeingRequest;
import ai.badmonkey.agentspaces.partybus.domain.Trip.PointOfInterest;
import ai.badmonkey.agentspaces.partybus.domain.Trip.TravelBrief;
import ai.badmonkey.agentspaces.partybus.fleet.PartyBusPlanner;
import ai.badmonkey.agentspaces.partybus.sources.Sources;
import com.embabel.agent.api.annotation.AchievesGoal;
import com.embabel.agent.api.annotation.Action;
import com.embabel.agent.api.annotation.Agent;
import com.embabel.agent.config.annotation.EnableAgents;
import com.embabel.agent.core.AgentPlatform;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;

import java.time.Duration;

/**
 * Party Bus as a Tripper-style Embabel application. This source set compiles
 * only under the {@code embabel} Maven profile, against the same Embabel
 * starter Tripper uses, and it is the part of the project that cannot be
 * verified in a network-restricted build; treat it as the shape of the
 * integration and run it from a laptop with Maven Central and the API keys.
 *
 * <p>Two things happen here. First, the bus boards inside the Spring Boot
 * application, exactly as the standalone demo boards it. Second, the fleet's
 * AgentCards come back in as planner actions: {@link PartyBusPlanner#bridge}
 * generates an {@code @Agent} with one {@code @Action} per remote card and
 * {@code deployTo(platform)} hands it to Embabel's {@code AgentPlatform}, so
 * the GOAP planner sees the sightseeing scout, the review reader, and every
 * other specialist as actions it can chain. The local {@link PartyBusPlannerAgent}
 * supplies the two actions the fleet cannot: turning a brief into the first
 * request, and turning the top sight into a review request. The goal, a
 * {@code ReviewDigest} from a {@code TravelBrief}, is reachable only through
 * the fleet, which is the whole point.
 */
@SpringBootApplication
@EnableAgents
public class PartyBusEmbabelApplication {

    /**
     * Runs the application.
     *
     * @param args passed to Spring Boot
     */
    public static void main(String[] args) {
        SpringApplication.run(PartyBusEmbabelApplication.class, args);
    }

    /** Boards the bus and bridges its cards into the platform once Spring is up. */
    @Bean
    CommandLineRunner boardTheBus(AgentPlatform platform) {
        return args -> {
            Sources sources = Sources.fromEnvironment();
            PartyBus.Seating bus = PartyBus.board(sources, 7700, 7590);
            RemoteActions remote = PartyBusPlanner.remoteActions(bus.plannerPeer());
            PartyBusPlanner.awaitActions(remote, 12, Duration.ofSeconds(30));
            EmbabelRemoteActions bridge = PartyBusPlanner.bridge(remote);
            if (!bridge.deployTo(platform)) {
                throw new IllegalStateException("the fleet agent did not deploy; are the cards in?");
            }
        };
    }

    /**
     * The planner's own two actions. Everything between them is a remote scout
     * the bridge generated from an AgentCard.
     */
    @Agent(description = "Plans a trip with the Party Bus fleet's specialists")
    public static class PartyBusPlannerAgent {

        @Action
        public SightseeingRequest askForSights(TravelBrief brief) {
            return new SightseeingRequest(brief.tripId(), "planner sights " + brief.to(),
                    brief.to(), brief.brief(), brief.departureDate(), brief.returnDate(),
                    PartyBus.party(brief.tripId()).contribution());
        }

        @Action
        public ReviewRequest askForReviewsOfTheTopSight(SightseeingOptions sights) {
            PointOfInterest top = sights.pointsOfInterest().isEmpty()
                    ? new PointOfInterest("the old town", "", "", "", "")
                    : sights.pointsOfInterest().get(0);
            return new ReviewRequest(sights.tripId(), "planner reviews " + top.name(),
                    top.name(), top.location());
        }

        @AchievesGoal(description = "Know what travelers say about the trip's top sight")
        @Action
        public ReviewDigest done(ReviewDigest digest) {
            return digest;
        }
    }
}

<profile>
      <id>embabel</id>
      <dependencyManagement>
        <dependencies>
          <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-dependencies</artifactId>
            <version>${spring-boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
          </dependency>
        </dependencies>
      </dependencyManagement>
      <dependencies>
        <dependency>
          <groupId>com.embabel.agent</groupId>
          <artifactId>embabel-agent-starter-openai</artifactId>
          <version>${embabel-agent.version}</version>
        </dependency>
        <dependency>
          <groupId>ai.badmonkey.agentspaces</groupId>
          <artifactId>agentspaces-spring-boot-starter</artifactId>
          <version>${agentspaces.version}</version>
        </dependency>
      </dependencies>
      <build>
        <plugins>
          <plugin>
            <groupId>org.codehaus.mojo</groupId>
            <artifactId>build-helper-maven-plugin</artifactId>
            <version>3.4.0</version>
            <executions>
              <execution>
                <id>add-embabel-sources</id>
                <phase>generate-sources</phase>
                <goals>
                  <goal>add-source</goal>
                </goals>
                <configuration>
                  <sources>
                    <source>src/embabel/java</source>
                  </sources>
                </configuration>
              </execution>
            </executions>
          </plugin>
        </plugins>
      </build>
    </profile>

# Party Bus under Spring Boot (the embabel profile). The standalone demo reads
# none of this; it boards the bus in code. With -Pembabel, the AgentSpaces
# starter and the Embabel starter both read it.
server:
  port: 8749            # Tripper flies 8747; the bus parks two doors down

agentspaces:
  keystore: ./partybus-keys
  bind: 0.0.0.0:7700
  transport:
    channel-auth: attested        # SPEC §5.6: TLS vouches for the sender, frames travel bare
    tls:
      enabled: true
  groups:
    - name: partybus
      founding: partybus-v1
      spaces:
        - { name: trip }
        - { name: research }
        - { name: signals }
        - { name: advisories }
        - { name: decisions }
        - { name: slots, strategy: AUCTION }
        - { name: bookings }
        - { name: fleet-control }
        - { name: c2-audit }
  embabel:
    remote-actions-enabled: true
    remote-action-timeout-millis: 60000
    remote-agent-name: partyBusFleet
  console:
    enabled: true
    port: 7590
    fleet-name: partybus
    command:
      enabled: true
      token: ${PARTYBUS_OPERATOR_TOKEN:partybus-operator}

embabel:
  models:
    default-llm: gpt-4.1-mini     # as Tripper's researcher; the plan itself deserves a better one

public static Turret over(PartyBusFleet.Peer peer) {
    Objects.requireNonNull(peer, "peer");
    ConsoleView.Builder view = ConsoleView.builder()
            .space(PartyBusFleet.RESEARCH, peer.space(PartyBusFleet.RESEARCH),
                    Research.FlightRequest.class, Research.HotelRequest.class,
                    Research.CarRequest.class, Research.SightseeingRequest.class,
                    Research.RestaurantRequest.class, Research.DietRequest.class,
                    Research.FoodieRequest.class, Research.PhotoRequest.class,
                    Research.ReviewRequest.class, Research.AccessibilityRequest.class,
                    Research.HospitalRequest.class, Research.NextDestinationRequest.class,
                    Safety.NewsSweep.class, Safety.VideoSweep.class, Safety.SocialSweep.class)
            .space(PartyBusFleet.SLOTS, peer.space(PartyBusFleet.SLOTS), DaySlot.class)
            .results(PartyBusFleet.SLOTS, ScheduledActivity.class,
                    a -> ((ScheduledActivity) a).plannedBy())
            .space(PartyBusFleet.BOOKINGS, peer.space(PartyBusFleet.BOOKINGS),
                    BookingOrder.class)
            .results(PartyBusFleet.BOOKINGS, BookingConfirmation.class,
                    c -> ((BookingConfirmation) c).clerk())
            .members(() -> peer.runtime().membership().allMembers())
            .discovery(peer.discovery());
    for (Class<?> result : RESULT_TYPES) {
        view.results(PartyBusFleet.RESEARCH, result, Turret::authorOf);
    }
    ConsoleView built = view.build();

    FleetCommander commander = FleetCommander.builder(peer.spaces()::get)
            .controlSpace(PartyBusFleet.CONTROL)
            .auditSpace(PartyBusFleet.AUDIT)
            .view(built)
            .command(startTrip())
            .command(throwEvent())
            .command(reroute(peer.space(PartyBusFleet.TRIP)))
            .build();

    FleetConsoleServer server = FleetConsoleServer.builder(built)
            .fleetName("partybus")
            .commandAndControl(commander, OPERATOR_TOKEN)
            .panel(itineraryPanel(peer))
            .panel(safetyPanel(peer))
            .panel(bookingsPanel(peer))
            .panel(energyPanel(peer))
            .panel(responsesPanel(peer))
            .panel(planPanel(peer))
            .build();
    return new Turret(peer, built, server);
}

static ConsoleCommand throwEvent() {
    return new ConsoleCommand() {
        @Override
        public String id() {
            return "throw-event";
        }

        @Override
        public String title() {
            return "Throw an event";
        }

        @Override
        public String description() {
            return "Storm, strike, protest, illness, or a lost passport; the fleet reacts.";
        }

        @Override
        public List<Field> fields() {
            return List.of(Field.text("tripId", "Trip id"), Field.text("date", "Date (ISO)"),
                    Field.select("kind", "Kind", Arrays.stream(EventKind.values())
                            .map(Enum::name).toList()),
                    Field.text("location", "Location"), Field.text("detail", "What happened"),
                    Field.number("severity", "Severity 1-5"));
        }

        @Override
        public Outcome execute(Map<String, String> args, CommandContext ctx) {
            String eventId = args.get("tripId") + "-op-" + System.currentTimeMillis();
            String entryId = ctx.dispatch(PartyBusFleet.TRIP, new TripEvent(args.get("tripId"),
                    args.get("date"), EventKind.valueOf(args.get("kind")),
                    args.get("location"), args.get("detail"),
                    (int) Double.parseDouble(args.getOrDefault("severity", "3")), eventId),
                    Duration.ofHours(12));
            ctx.audit("throw-event", args.get("kind") + " at " + args.get("location"));
            return Outcome.dispatched(args.get("kind") + " thrown", entryId);
        }
    };
}

static ConsolePanel itineraryPanel(PartyBusFleet.Peer peer) {
    Space slots = peer.space(PartyBusFleet.SLOTS);
    return panel("itinerary", "Itinerary (auction winners)", () -> {
        Map<String, ScheduledActivity> byDay = new java.util.TreeMap<>();
        for (ScheduledActivity a : slots.readAll(Template.of(ScheduledActivity.class), 200)) {
            ScheduledActivity current = byDay.get(a.date());
            if (current == null || a.round() > current.round()) {
                byDay.put(a.date(), a);
            }
        }
        List<Map<String, Object>> rows = new ArrayList<>();
        for (ScheduledActivity a : byDay.values()) {
            rows.add(row("date", a.date(), "kind", a.kind(), "title", a.title(),
                    "plannedBy", a.plannedBy(), "bid", a.costUsd(), "round", a.round()));
        }
        return rows;
    });
}

public static List<Scene> lisbonWeek() {
    return List.of(
            new Scene(0, EventKind.CALM, "Wheels down. Pastel de nata count: 3 before the hotel.", 1),
            new Scene(1, EventKind.WEATHER, "Atlantic squall parks over the city until mid-afternoon.", 2),
            new Scene(2, EventKind.ILLNESS, "Marco ordered the clams. Marco regrets the clams.", 3),
            new Scene(3, EventKind.PROTEST, "Transport unions march on the Assembleia; Baixa is sealed off.", 3),
            new Scene(4, EventKind.STRIKE, "Metro and ferry strike; the tram to Belém is a rumor.", 3),
            new Scene(5, EventKind.CALM, "Sun, a rooftop, and nothing on fire.", 1),
            new Scene(6, EventKind.LOST_PASSPORT, "Priya's passport is in a taxi. Somewhere.", 4));
}

/**
 * Lives the trip, day by day, on the calling thread.
 *
 * @param brief the brief that defines the days
 * @param script the scenes to play
 * @return one narration per day, in order
 * @throws InterruptedException if interrupted between days
 */
public List<Narration> live(TravelBrief brief, List<Scene> script) throws InterruptedException {
    List<Narration> story = new ArrayList<>();
    List<String> days = brief.days();
    String pace = "EASY";
    for (int i = 0; i < days.size(); i++) {
        String date = days.get(i);
        // The day's event lands first, so the travelers wake up to it.
        TripEvent event = null;
        for (Scene scene : script) {
            if (scene.dayIndex() == i && scene.kind() != EventKind.CALM) {
                event = new TripEvent(brief.tripId(), date, scene.kind(), brief.to(),
                        scene.detail(), scene.severity(), brief.tripId() + "-d" + i + "-"
                        + scene.kind().name().toLowerCase());
                trip.write(event, Lease.of(Duration.ofHours(12)));
            }
        }
        Thread.sleep(300);
        trip.write(new TripDay(brief.tripId(), date, i, brief.to(), pace),
                Lease.of(Duration.ofHours(12)));
        Thread.sleep(dayLength.toMillis());

        OptionalDouble energy = aggregate.estimate(Itinerary.energyEpoch(brief.tripId(), date));
        if (energy.isEmpty()) {
            // No share reached this peer yet: the reports in the space are the fallback.
            energy = trip.readAll(Template.of(EnergyReport.class)
                            .where("tripId", eq(brief.tripId())).where("date", eq(date)), 20)
                    .stream().mapToInt(EnergyReport::energy).average();
        }
        List<ScheduledActivity> schedule = slots.readAll(Template.of(ScheduledActivity.class)
                .where("tripId", eq(brief.tripId())).where("date", eq(date)), 10);
        story.add(new Narration(date, pace, energy, event, schedule));
        narrator.accept(story.get(story.size() - 1));

        String next = paceFor(energy);
        if (i + 1 < days.size() && next.equals("REST")) {
            reauction(brief, days.get(i + 1), next);
        }
        pace = next;
    }
    return story;
}

public static String paceFor(OptionalDouble energy) {
    if (energy.isEmpty()) {
        return "FULL";
    }
    if (energy.getAsDouble() < 40) {
        return "REST";
    }
    return energy.getAsDouble() < 60 ? "EASY" : "FULL";
}

/**
 * Boards the bus: starts every peer, binds every agent, forms the booking
 * quorum, and starts the turret.
 *
 * @param sources     the specialists' sources
 * @param basePort    the first TCP port; peers take consecutive ports
 * @param consolePort the turret's HTTP port (0 for any)
 * @return the seating
 * @throws Exception on startup failure
 */
public static Seating board(Sources sources, int basePort, int consolePort) throws Exception {
    List<Peer> peers = new ArrayList<>();
    int[] port = {basePort};
    Function<String, Peer> seat = name -> {
        try {
            Peer peer = PartyBusFleet.startPeer(name, port[0]++,
                    peers.isEmpty() ? 0 : basePort);
            peers.add(peer);
            return peer;
        } catch (Exception e) {
            throw new IllegalStateException("cannot seat " + name, e);
        }
    };

    Peer turretPeer = seat.apply("turret");

    // The concierge desk: the agents that watch the whole trip rather than one task.
    Peer concierge = seat.apply("concierge");
    concierge.bind(new Dispatcher());
    concierge.bind(new Assemblers.StayFinder());
    concierge.bind(new Assemblers.PlanAssembler(sources));
    concierge.bind(new SafetyPanel.Lead("safety-lead"));

    // The scouts and advisors: the research bag's workers, four to a peer.
    Peer scoutsA = seat.apply("scouts-a");
    scoutsA.bind(new Scouts.FlightScout(sources));
    scoutsA.bind(new Scouts.HotelScout(sources));
    scoutsA.bind(new Scouts.CarScout(sources));
    scoutsA.bind(new Scouts.SightseeingScout(sources));
    Peer scoutsB = seat.apply("scouts-b");
    scoutsB.bind(new Scouts.RestaurantScout(sources));
    scoutsB.bind(new Scouts.PhotoCurator(sources));
    scoutsB.bind(new Scouts.ReviewReader(sources));
    scoutsB.bind(new Scouts.NextDestinationAdvisor(sources));
    Peer advisors = seat.apply("advisors");
    advisors.bind(new Advisors.DietAdvisor(sources));
    advisors.bind(new Advisors.FoodieAdvisor(sources));
    advisors.bind(new Advisors.AccessibilityAdvisor(sources));
    advisors.bind(new Advisors.HospitalLocator(sources));

    // The safety pipeline: watchers, the analyst, and the three-seat panel.
    Peer watchers = seat.apply("watchers");
    watchers.bind(new Watchers.NewsWatcher(sources));
    watchers.bind(new Watchers.VideoWatcher(sources));
    watchers.bind(new Watchers.SocialWatcher(sources));
    watchers.bind(new Watchers.SentimentAnalyst(sources));
    // One peer per panelist: a ballot's voter is the peer's own agent, so
    // three seats sharing a peer would be one voter and the three-seat quorum
    // could never close (QA4 A4-4).
    for (String seatName : SafetyPanel.SEATS) {
        Peer panelist = seat.apply("panelist-" + seatName);
        panelist.bind(new SafetyPanel.Panelist(seatName));
    }

    // The activity auction: one bid function per peer, so one planner per peer.
    seat.apply("dining-planner").bind(new Planners.DiningPlanner());
    seat.apply("sights-planner").bind(new Planners.SightsPlanner());
    seat.apply("photo-walk-planner").bind(new Planners.PhotoWalkPlanner());
    seat.apply("rest-planner").bind(new Planners.RestPlanner());

    // The party itself: one peer per traveler, so each carries its own ballot
    // and the party quorum is a quorum of members (QA4 A4-4).
    Travelers party = party("lisbon-2026");
    int[] stamina = {85, 70, 55};
    for (int i = 0; i < party.travelers().size(); i++) {
        Traveler traveler = party.travelers().get(i);
        Peer seatOf = seat.apply("traveler-" + traveler.name().toLowerCase());
        seatOf.bind(new Party.TravelerAgent(traveler, stamina[i], seatOf.aggregate()));
    }
    Peer bus = seat.apply("the-bus");
    bus.bind(new Party.Lead(party.travelers().size()));

    // The booking desk: three clerks, one Raft quorum.
    List<Peer> clerks = List.of(seat.apply("clerk-a"), seat.apply("clerk-b"),
            seat.apply("clerk-c"));
    PartyBusFleet.formBookingQuorum(clerks);

    // The planner is where Tripper's own agent would run; the simulator rides with the turret.
    Peer plannerPeer = seat.apply("planner");
    Peer simulatorPeer = turretPeer;

    Turret turret = Turret.over(turretPeer);
    turretPeer.run(turret);
    int bound = turret.start(consolePort);
    System.out.println("turret: http://localhost:" + bound + "/  (operator token: "
            + Turret.OPERATOR_TOKEN + ")");
    return new Seating(peers, turretPeer, plannerPeer, simulatorPeer);
}