Skip to main content

cratefield_core/
harness.rs

1//! The `Harness` builder and axum router assembly (issue #2, architecture
2//! section 4).
3//!
4//! `Harness::build()` collects **all** configuration problems and reports
5//! them together; `Harness::router(ports)` mounts every module under
6//! `/v1/<name>` and adds `GET /__health` and `GET /__ready` plus the
7//! middleware stack from architecture section 6.
8
9use std::collections::BTreeSet;
10use std::collections::HashMap;
11use std::sync::Arc;
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::time::Duration;
14
15use axum::Router;
16use axum::extract::{DefaultBodyLimit, State};
17use axum::http::{HeaderMap, Method, StatusCode, header};
18use axum::middleware::{Next, from_fn_with_state};
19use axum::response::{IntoResponse, Response};
20use axum::routing::{get, post};
21use bytes::Bytes;
22use serde_json::json;
23use tracing::error;
24
25use crate::admin::require_admin;
26use crate::config::Config;
27use crate::config::ConfigError;
28use crate::events::EventBus;
29use crate::http::{
30    Json, MAX_BODY_BYTES, ScopeState, cors_layer, rate_limited, scope_layer,
31    security_headers_layer, token_response_layer,
32};
33use crate::module::{HARNESS_API, Module, ModuleContext, harness_api_mismatch};
34use crate::ports::Dispatcher;
35use crate::ports::{
36    Clock, Database, Port, Ports, RateLimiter, Statement, SystemClock, warn_undeclared_ports,
37};
38use crate::problem::Problem;
39use crate::problems::SLUGS;
40use crate::rate_limit::{RateLimit, RateLimitFailure};
41use crate::route_policy::deployed_env;
42use crate::scheduled::ScheduledBudget;
43use crate::scope::Scope;
44use crate::sidecar::{
45    GatewayGuard, SIDECAR_REQUIRE_GATEWAY, SidecarMount, X_HARNESS_GATEWAY, gateway_guard,
46    gateway_signer, mint_gateway, truthy,
47};
48use crate::signer::HmacSigner;
49use crate::stream::{RequestStream, StreamRoute};
50use crate::surface::{
51    MAX_SIDECAR_SURFACE_BYTES, RenderedSurface, SurfaceDocument, SurfaceSource, UiContext, UiMount,
52    sanitize_sidecar_document, strip_unguarded_captcha_actions,
53};
54use crate::template::{Template, TemplateRegistry};
55use crate::venture::{Venture, VentureEnv};
56
57/// The two checks about a module's tables: that nobody else claimed one, and
58/// that its personal-data declarations can mean something.
59///
60/// Extracted from `HarnessBuilder::build`, which sat one line under the
61/// workspace's function-length lint before this. Collecting one more category
62/// of problem should not be what makes that function too long to follow, and
63/// the two checks read better together anyway: both are about the tables the
64/// module just claimed, and the second is only meaningful against the first.
65fn check_tables(
66    module: &dyn Module,
67    tables: &mut HashMap<&'static str, &'static str>,
68    errors: &mut ConfigError,
69) {
70    let name = module.name();
71    for table in module.tables() {
72        match tables.get(table) {
73            Some(owner) => errors.push(format!(
74                "duplicate table `{table}` claimed by modules `{owner}` and `{name}`"
75            )),
76            None => {
77                tables.insert(table, name);
78            }
79        }
80    }
81
82    // Against the tables the module just claimed, so a rename that misses a
83    // declaration is a build error rather than a row that quietly stops being
84    // exported.
85    let owns = module.tables();
86    let mut declared: Vec<&'static str> = Vec::new();
87    for set in module.personal_data() {
88        for problem in set.validate(name, owns) {
89            errors.push(problem);
90        }
91        if declared.contains(&set.table) {
92            errors.push(format!(
93                "module `{name}` declares table `{}` twice in personal_data()",
94                set.table
95            ));
96        }
97        declared.push(set.table);
98    }
99}
100
101/// A runtime resolves environment bindings into [`Ports`] and declares
102/// statically which ports it can provide, so `Harness::build` can reject a
103/// module that requires something the runtime will never hand it
104/// (ADR 0002). Reference implementation: `cratefield-runtime-cloudflare`.
105pub trait Runtime: Send + Sync + 'static {
106    fn provides(&self) -> Vec<Port>;
107    /// Whether the named port is not merely present but usable for its
108    /// production duty (issue #133): a captcha adapter that reports
109    /// itself unbound cannot protect a `HumanForm` route, so the runtime
110    /// that wraps it says "no" here even though `provides()` lists the
111    /// port. Default: presence in `provides()`.
112    fn effectively_configured(&self, port: Port) -> bool {
113        self.provides().contains(&port)
114    }
115}
116
117/// A built harness: immutable after `build()`.
118pub struct Harness {
119    venture: Arc<Venture>,
120    modules: Vec<Arc<dyn Module>>,
121    templates: Arc<TemplateRegistry>,
122    events: EventBus,
123    runtime: Option<Arc<dyn Runtime>>,
124    /// The single module-provided `/.well-known` router, if any
125    /// (issue #46); nested at the root by `router()`.
126    well_known: Option<Router>,
127    /// The composed UI surface (ADR 0010), rendered once for
128    /// `GET /__surface`: the admin variant and the public subset.
129    surface: Arc<SurfaceVariants>,
130    /// The renderer mounted at `/ui`, if the venture chose one.
131    ui: Option<Arc<dyn UiMount>>,
132    /// Every module's personal-data declarations, composed once at build so a
133    /// request does not walk the module list to answer an export.
134    personal_data: Arc<crate::PersonalDataCatalog>,
135    /// Whether the boot has already recorded the operator's
136    /// `HARNESS_ALLOW_UNPROTECTED_WRITES` acceptance (issue #143), and the
137    /// `HARNESS_ALLOW_UNLIMITED_PUBLIC_ROUTES` one (issue #437). Workers
138    /// rebuild the router per request, so the recording has to gate itself:
139    /// one line per waiver, the first time the gate serves on it — not one
140    /// per request. Per-`Harness` rather than a process static, which is the
141    /// same span in every real deployment (the harness is built once per
142    /// isolate) and keeps tests and multi-venture hosts from suppressing
143    /// each other's records.
144    unprotected_acceptance_recorded: AtomicBool,
145    unlimited_acceptance_recorded: AtomicBool,
146}
147
148struct SurfaceVariants {
149    document: Arc<SurfaceDocument>,
150    full: RenderedSurface,
151    public: RenderedSurface,
152}
153
154impl SurfaceVariants {
155    fn compose(
156        venture: &Venture,
157        modules: &[Arc<dyn Module>],
158        ui: Option<&Arc<dyn UiMount>>,
159    ) -> Self {
160        let mut document = SurfaceDocument::compose(venture, modules);
161        document.ui = ui.and_then(|ui| ui.describe());
162        Self {
163            full: RenderedSurface::render(&document),
164            public: RenderedSurface::render(&document.public()),
165            document: Arc::new(document),
166        }
167    }
168}
169
170impl std::fmt::Debug for Harness {
171    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
172        let modules: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
173        f.debug_struct("Harness")
174            .field("venture", &self.venture.name)
175            .field("modules", &modules)
176            .finish_non_exhaustive()
177    }
178}
179
180impl Harness {
181    /// Cheap shared handle: composition is `Arc` fields, so the only
182    /// sensible clone is the `Arc` itself. The reconciler hands one
183    /// `Arc` to every tenant task it fans out.
184    #[must_use]
185    pub fn clone_arc(self: &std::sync::Arc<Self>) -> std::sync::Arc<Self> {
186        std::sync::Arc::clone(self)
187    }
188    pub fn builder() -> HarnessBuilder {
189        HarnessBuilder::default()
190    }
191
192    pub fn venture(&self) -> &Arc<Venture> {
193        &self.venture
194    }
195
196    pub fn modules(&self) -> &[Arc<dyn Module>] {
197        &self.modules
198    }
199
200    /// The largest body a runtime may buffer for the route at `path`,
201    /// answerable before any byte of it has been read (issue #440).
202    ///
203    /// `DefaultBodyLimit` inside `router()` is the precise per-route
204    /// enforcer, but it fires only once the body is already resident — too
205    /// late for a runtime that must buffer into a fixed memory ceiling (a
206    /// Workers isolate). This is the coarse pre-buffer ceiling that runtime
207    /// consults instead: for a path of the form `/v1/<name>` or
208    /// `/v1/<name>/...` it is the named module's [`Module::max_body_bytes`],
209    /// and for everything else — an unknown module, `/ui/*`, `/__events`,
210    /// `/.well-known/*`, or any path outside `/v1` — [`MAX_BODY_BYTES`].
211    ///
212    /// The result is floored at [`MAX_BODY_BYTES`] even for the module
213    /// routes: a module may raise the ceiling for a route (LinkedIn's image
214    /// upload), but the runtime guard sits in front of the router, and a
215    /// module that tightened it below what the router itself accepts would
216    /// 413 working requests.
217    ///
218    /// `path` is the URL path only — what `url.path()` returns. A query
219    /// string is tolerated and ignored, and empty segments from a leading,
220    /// trailing or doubled slash are collapsed; `/v1` with no module
221    /// segment is simply the default ceiling.
222    pub fn max_body_bytes(&self, path: &str, cfg: &dyn Config) -> usize {
223        let Some((name, _)) = module_and_segments(path) else {
224            return MAX_BODY_BYTES;
225        };
226        self.modules
227            .iter()
228            .find(|module| module.name() == name)
229            .map_or(MAX_BODY_BYTES, |module| {
230                MAX_BODY_BYTES.max(module.max_body_bytes(cfg))
231            })
232    }
233
234    /// The streaming route that serves `method` at `path`, if the module
235    /// mounted there declared one — answering its own
236    /// [`StreamRoute::max_bytes`] ceiling (issue #585).
237    ///
238    /// The counterpart of [`Harness::max_body_bytes`] on the streaming side:
239    /// a runtime consults this **before** reading a body, to know whether
240    /// the route streams (and with what ceiling) or buffers, exactly as
241    /// `max_body_bytes` tells it how much a buffered route may hold. The
242    /// ceiling is returned verbatim — a streaming route's `max_bytes` is the
243    /// precise per-route limit, not a coarse pre-buffer guard, so it is not
244    /// floored at [`MAX_BODY_BYTES`].
245    ///
246    /// Path normalisation is exact, not [`Harness::max_body_bytes`]': a query
247    /// or fragment is ignored, but empty segments are **not** collapsed, so a
248    /// trailing or doubled slash matches no route — the request then keeps
249    /// its buffered body and axum's own 404, exactly as without streaming.
250    /// `path` is the URL path only; `method` is the request's.
251    pub fn streaming_route(&self, path: &str, method: &axum::http::Method) -> Option<usize> {
252        let (name, segments) = exact_module_and_segments(path)?;
253        let module = self.modules.iter().find(|module| module.name() == name)?;
254        match_streaming(module.streaming_routes(), method, &segments)
255    }
256
257    pub fn templates(&self) -> &Arc<TemplateRegistry> {
258        &self.templates
259    }
260
261    pub fn events(&self) -> &EventBus {
262        &self.events
263    }
264
265    /// Builds the context a module sees: its declared ports (view), the
266    /// config, the shared bus, templates and venture. `router()` uses this
267    /// per module; `cratefield-runtime-cloudflare` uses it for scheduled
268    /// fan-out.
269    pub fn module_context(&self, module: &dyn Module, ports: &Ports) -> ModuleContext {
270        // `venture.env` here is the environment the **deployment**
271        // declares, not the compiled default (issue #143).
272        //
273        // #143 corrected the boot gate and the surface merge and left
274        // this path alone, which is the one that decides per request. A
275        // module reading `ctx.venture.env` was getting `Development` on a
276        // Worker serving production — and `module-waitlist` reads exactly
277        // that to decide whether a missing `Captcha` port fails closed.
278        // So the rule that says "no captcha port in production is a
279        // refusal" never fired for the venture it was written for.
280        //
281        // Corrected here rather than at the call site so the field cannot
282        // lie: leaving `ctx.venture.env` looking right and being wrong is
283        // how this survived the fix that was meant to remove it.
284        let venture = self.venture_as_deployed(ports.config.as_ref());
285        ModuleContext {
286            config: Arc::clone(&ports.config),
287            ports: ports.view_for(module),
288            events: self.events_for(ports),
289            templates: Arc::clone(&self.templates),
290            venture,
291            unprotected_writes_accepted: crate::route_policy::unprotected_writes_override(
292                ports.config.as_ref(),
293            )
294            .is_some(),
295            personal_data: Arc::clone(&self.personal_data),
296            ui_mounted: self.ui.is_some(),
297            // Unbounded: the core has no invocation limits to spend. The
298            // runtimes replace this per module during scheduled fan-out
299            // (issue #537); request handlers keep this one.
300            scheduled: Arc::new(ScheduledBudget::unbounded()),
301        }
302    }
303
304    /// The sidecar mounts that apply: read from configuration, not from
305    /// the composition (ADR 0009), so the same artifact serves ventures
306    /// with and without them. A malformed table mounts nothing and is
307    /// logged; a mount that collides with an in-process module is
308    /// dropped and logged. Neither takes down the in-process modules.
309    /// This deployment's sidecar-role enforcement state (issue #131):
310    /// whether the gate is closed, the key that verifies a stamp, and the
311    /// sidecar's own admin token — which never crosses the boundary and is
312    /// only ever re-asserted for a request the host already authorized.
313    fn gateway_guard_state(ports: &Ports, gateway: Option<&Arc<HmacSigner>>) -> GatewayGuard {
314        GatewayGuard {
315            require: truthy(ports.config.as_ref(), SIDECAR_REQUIRE_GATEWAY),
316            signer: gateway.map(Arc::clone),
317            admin_token: ports
318                .config
319                .get("ADMIN_TOKEN")
320                .filter(|token| !token.is_empty()),
321        }
322    }
323
324    /// The surface this deployment serves: the composed one, plus any
325    /// mounted sidecar's public part merged in per request (issue #131).
326    fn merged_surface(
327        &self,
328        mounts: Vec<SidecarMount>,
329        ports: &Ports,
330        gateway: Option<Arc<HmacSigner>>,
331        env: VentureEnv,
332    ) -> Arc<dyn SurfaceSource> {
333        Arc::new(MergedSurface {
334            base: Arc::clone(&self.surface),
335            mounts,
336            dispatcher: ports.dispatcher.clone(),
337            gateway,
338            env,
339            captcha_configured: ports.captcha.is_some(),
340        })
341    }
342
343    /// The venture descriptor with the environment the **deployment**
344    /// declares (issue #143), which is what anything reporting or
345    /// enforcing on it must read. Returns the original `Arc` untouched
346    /// when the two already agree, which is every venture that sets its
347    /// environment honestly in code.
348    fn venture_as_deployed(&self, config: &dyn Config) -> Arc<Venture> {
349        let env = deployed_env(self.venture.env, config);
350        if env == self.venture.env {
351            return Arc::clone(&self.venture);
352        }
353        let mut deployed = (*self.venture).clone();
354        deployed.env = env;
355        Arc::new(deployed)
356    }
357
358    /// Nests each in-process module under its `/v1/<name>` prefix.
359    /// Split out of [`Harness::router`] so that method stays readable.
360    /// Layered in [`Harness::router`], not here: a layer applied inside
361    /// this function would wrap only the module routes, and the sidecar
362    /// prefixes nested after it would sit outside the admin floor.
363    fn nest_modules(&self, ports: &Ports) -> Router {
364        let mut api = Router::new();
365        for module in &self.modules {
366            let ctx = self.module_context(module.as_ref(), ports);
367            api = api.nest(&format!("/v1/{}", module.name()), module.router(ctx));
368        }
369        api
370    }
371
372    /// The production readiness verdict for the environment this
373    /// deployment actually runs in (issue #143).
374    ///
375    /// `HarnessBuilder::build` already ran this against the **compiled**
376    /// `Venture::env`. When the deployment declares a stricter one that
377    /// check was made against the wrong environment, so it is re-made
378    /// here against the same runtime report — the answer must not depend
379    /// on whether anyone remembered to call `.env()`.
380    ///
381    /// The limiter leg is decided here against the **resolved** port, not
382    /// the runtime's advertisement (issue #437): a Cloudflare binding that
383    /// fails to resolve degrades to `ports.rate_limiter == None`, and this
384    /// is the check that catches it. The signer leg likewise (issue #478):
385    /// a harness configuration that fails to parse leaves `ports.signer ==
386    /// None` behind an advertised `Port::Signer`.
387    fn production_readiness_now(
388        &self,
389        env: VentureEnv,
390        config: &dyn Config,
391        rate_limiter_ready: bool,
392        signer_ready: bool,
393    ) -> Vec<String> {
394        if let Some(note) = crate::route_policy::env_disagreement(self.venture.env, env) {
395            tracing::warn!("{note}");
396            crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &note);
397        }
398        // Each escape hatch is read here — in the serving path — and
399        // passed to its own leg, so a waiver covers exactly the control it
400        // names and refuses alongside whatever remains unwaived.
401        let unprotected = crate::route_policy::unprotected_writes_override(config);
402        let unlimited = crate::route_policy::unlimited_public_routes_override(config);
403        let guards = crate::route_policy::WriteGuards::collect(&self.modules);
404        let mut problems = crate::route_policy::production_readiness(
405            env,
406            &guards,
407            self.runtime.as_ref(),
408            rate_limiter_ready,
409            signer_ready,
410            unprotected.as_deref(),
411            unlimited.as_deref(),
412        );
413        // The webhook-secret leg is boot-time-only (issue #533): it reads
414        // the deployment config, which the build-time gate does not have.
415        // It has no waiver, so it is held out of the loop below — that loop
416        // names the escape hatch, which clears nothing here.
417        let hmac_problems = crate::route_policy::webhook_secret_readiness(env, &guards, config);
418        if problems.is_empty() && hmac_problems.is_empty() {
419            // An operator may accept this deployment's gaps explicitly,
420            // and the acceptance is recorded rather than discarded
421            // (issue #143) — once, even though Workers rebuild this
422            // router per request (issue #437).
423            // What the gate WOULD have refused, asked for again with no
424            // waivers: the waived text is never pushed, so this is the only
425            // way to name the problems the operator accepted — and #441's
426            // record is only accountable if it carries them.
427            let accepted = crate::route_policy::production_readiness(
428                env,
429                &guards,
430                self.runtime.as_ref(),
431                rate_limiter_ready,
432                signer_ready,
433                None,
434                None,
435            );
436            self.record_acceptances_once(
437                &guards,
438                rate_limiter_ready,
439                unprotected.as_deref(),
440                unlimited.as_deref(),
441                &accepted,
442            );
443            return problems;
444        }
445        // No blanket escape hatch here any more. Each waiver is passed
446        // into `production_readiness` above and applied to the leg it
447        // names, so what comes back is already only the UNWAIVED problems
448        // — clearing them all on `unprotected_writes_override` would let a
449        // waiver for one control excuse another, which is exactly what
450        // `a_refusing_deployment_records_no_acceptance_at_all` forbids: a
451        // captcha waiver must not serve a venture with no rate limiter.
452        for problem in &problems {
453            let detail = format!(
454                "refusing guarded routes: {problem} — set {} to a reason to accept this \
455                 explicitly while the port is wired",
456                crate::route_policy::ALLOW_UNPROTECTED_WRITES
457            );
458            error!("{detail}");
459            crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
460        }
461        problems.extend(hmac_problems);
462        problems
463    }
464
465    /// Records an operator's escape-hatch acceptance exactly once, next to
466    /// the boot decision that served on it (issue #143, #437). Each guard
467    /// below is the readiness leg it mirrors minus the waiver clause, so a
468    /// waiver is recorded only when it actually removed a refusal — not
469    /// when the port was fine all along.
470    fn record_acceptances_once(
471        &self,
472        guards: &crate::route_policy::WriteGuards,
473        rate_limiter_ready: bool,
474        unprotected: Option<&str>,
475        unlimited: Option<&str>,
476        accepted: &[String],
477    ) {
478        if guards.needs_captcha()
479            && !crate::route_policy::captcha_effective(self.runtime.as_ref())
480            && let Some(reason) = unprotected
481            && !self
482                .unprotected_acceptance_recorded
483                .swap(true, Ordering::Relaxed)
484        {
485            let detail = format!(
486                "serving guarded routes unprotected on an operator's recorded acceptance \
487                 (reason: {reason}; problems: {})",
488                accepted.join("; ")
489            );
490            tracing::warn!(
491                control = "production-readiness",
492                acceptance = crate::route_policy::ALLOW_UNPROTECTED_WRITES,
493                reason,
494                "serving captcha-guarded routes unprotected on an operator's recorded acceptance"
495            );
496            // Forwarded, not just traced (issue #441): on wasm the tracing
497            // event goes nowhere, and an acceptance nobody can read is an
498            // acceptance nobody answers for.
499            crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
500        }
501        if guards.needs_rate_limiter()
502            && !rate_limiter_ready
503            && let Some(reason) = unlimited
504            && !self
505                .unlimited_acceptance_recorded
506                .swap(true, Ordering::Relaxed)
507        {
508            let detail = format!(
509                "serving public writes and admin routes without a rate limiter on an \
510                 operator's recorded acceptance (reason: {reason}; problems: {})",
511                accepted.join("; ")
512            );
513            tracing::warn!(
514                control = "production-readiness",
515                acceptance = crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES,
516                reason,
517                "serving public writes and admin routes without a rate limiter on an \
518                 operator's recorded acceptance"
519            );
520            // Forwarded for the same reason as the leg above (issue #441).
521            crate::logging::forward_control_event(crate::logging::ControlLevel::Warn, &detail);
522        }
523    }
524
525    /// Nests one forwarding router per mounted sidecar (issue #131).
526    /// Split out of [`Harness::router`] so that method stays readable.
527    fn nest_sidecars(
528        mut api: Router,
529        mounted: &[SidecarMount],
530        ports: &Ports,
531        gateway: Option<&Arc<HmacSigner>>,
532    ) -> Router {
533        for mount in mounted {
534            api = api.nest(
535                &format!("/v1/{}", mount.name),
536                crate::sidecar::router(
537                    mount.clone(),
538                    ports.dispatcher.clone(),
539                    Arc::clone(&ports.config),
540                    ports.rate_limiter.clone(),
541                    gateway.map(Arc::clone),
542                ),
543            );
544        }
545        api
546    }
547
548    fn sidecar_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
549        let mounts = match crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()) {
550            Ok(mounts) => mounts,
551            Err(errors) => {
552                for error in errors {
553                    let detail = format!("ignoring the sidecar mount table: {error}");
554                    tracing::error!(error, "ignoring the sidecar mount table");
555                    // A dropped mount quietly unmounts a service (issue
556                    // #441): the operator must be able to read why.
557                    crate::logging::forward_control_event(
558                        crate::logging::ControlLevel::Error,
559                        &detail,
560                    );
561                }
562                crate::sidecar::SidecarMounts::default()
563            }
564        };
565        let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
566        for collision in mounts.collisions(&module_names) {
567            let detail = format!("ignoring the colliding sidecar mount: {collision}");
568            tracing::error!(error = collision, "ignoring the colliding sidecar mount");
569            crate::logging::forward_control_event(crate::logging::ControlLevel::Error, &detail);
570        }
571        mounts
572            .iter()
573            .filter(|m| !module_names.contains(&m.name.as_str()))
574            .cloned()
575            .collect()
576    }
577
578    /// [`Harness::sidecar_mounts`] without the logging: `events_for` runs
579    /// per module per router, and re-logging the same malformed table once
580    /// per module would turn one operator error into a page of noise.
581    fn resolved_mounts(&self, ports: &Ports) -> Vec<SidecarMount> {
582        let mounts =
583            crate::sidecar::SidecarMounts::from_config(ports.config.as_ref()).unwrap_or_default();
584        let module_names: Vec<&str> = self.modules.iter().map(|m| m.name()).collect();
585        mounts
586            .iter()
587            .filter(|m| !module_names.contains(&m.name.as_str()))
588            .cloned()
589            .collect()
590    }
591
592    /// The bus a module sees: the handlers collected at `build`, plus —
593    /// when this deployment mounts sidecars — the forwarder that carries
594    /// every emission across the boundary (issue #62, ADR 0017). Built
595    /// here rather than in `build` because only `router(ports)` has the
596    /// resolved `Dispatcher` and mount table; that is also the choice that
597    /// keeps `Scope` and every handler signature untouched, so the
598    /// contract version does not move.
599    fn events_for(&self, ports: &Ports) -> EventBus {
600        let mounts = self.resolved_mounts(ports);
601        if mounts.is_empty() {
602            return self.events.clone();
603        }
604        let Some(dispatcher) = ports.dispatcher.clone() else {
605            return self.events.clone();
606        };
607        let gateway = gateway_signer(
608            ports.config.as_ref(),
609            ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
610        );
611        self.events
612            .clone()
613            .forwarding_to(crate::sidecar::EventForwarder::new(
614                mounts, dispatcher, gateway,
615            ))
616    }
617
618    /// The venture's UI plane, if it declared one. Lifted out of
619    /// [`Harness::router`] only because adding `/__events` to that function
620    /// put it over the line limit, and a mechanical extraction is a better
621    /// answer than an `allow` on the lint that noticed.
622    fn ui_router(
623        &self,
624        api: &Router,
625        ports: &Ports,
626        surface: &Arc<dyn SurfaceSource>,
627    ) -> Option<Router> {
628        self.ui.as_ref().map(|ui| {
629            ui.router(UiContext {
630                surface: Arc::clone(surface),
631                api: api.clone(),
632                config: Arc::clone(&ports.config),
633                venture: Arc::clone(&self.venture),
634                captcha_configured: ports.captcha.is_some(),
635                signer: ports.signer.clone(),
636                rate_limiter: ports.rate_limiter.clone(),
637            })
638            .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
639        })
640    }
641
642    /// The root router: the probes and the surface, inbound events, then
643    /// the API, `/.well-known` and the UI plane nested on top. A sibling
644    /// of [`Harness::ui_router`] in the same spirit — the assembly is one
645    /// concern, [`Harness::router`] another, and keeping them apart is
646    /// what lets each stay readable.
647    fn root_router(
648        &self,
649        api: Router,
650        ui: Option<Router>,
651        gateway: Option<Arc<HmacSigner>>,
652        health_state: HealthState,
653        ready_state: ReadyState,
654        surface_state: SurfaceState,
655    ) -> Router {
656        let root = Router::new()
657            .route("/__health", get(health_handler))
658            .with_state(health_state)
659            .route("/__ready", get(ready_handler))
660            .with_state(ready_state)
661            .route("/__surface", get(surface_handler))
662            .with_state(surface_state)
663            // Empty without a gateway secret, and merging an empty router
664            // adds no route — see `inbound_events_route` for why absent.
665            .merge(self.inbound_events_route(gateway).unwrap_or_default())
666            .merge(api);
667        let root = match &self.well_known {
668            Some(well_known) => root.nest(
669                "/.well-known",
670                well_known
671                    .clone()
672                    .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
673            ),
674            None => root,
675        };
676        match ui {
677            Some(ui) => root.nest("/ui", ui),
678            None => root,
679        }
680    }
681
682    /// The sidecar half of event forwarding (issue #62, ADR 0017):
683    /// `POST /__events`, delivering into this deployment's **own** bus.
684    ///
685    /// Deliberately not the forwarding bus from [`Harness::events_for`]. A
686    /// deployment can be both a host and, to someone above it, a sidecar;
687    /// delivering an inbound event through a bus that forwards would re-post
688    /// it to this deployment's own mounts, and two deployments that mount
689    /// each other would loop with nothing able to detect it.
690    ///
691    /// `None` without a gateway secret: absent is safer than open. Without
692    /// the shared secret there is no way to tell the host's forward from
693    /// anyone else's `POST`, and an unauthenticated event trigger would let
694    /// a stranger forge the payloads in-process handlers act on.
695    fn inbound_events_route(&self, gateway: Option<Arc<HmacSigner>>) -> Option<Router> {
696        gateway.as_ref()?;
697        Some(
698            Router::new()
699                .route(
700                    "/__events",
701                    post(crate::sidecar::events_inbound)
702                        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES)),
703                )
704                .with_state(Arc::new(crate::sidecar::InboundEvents {
705                    bus: self.events.clone(),
706                    gateway,
707                })),
708        )
709    }
710
711    /// The runtime this harness was validated against, if one was supplied.
712    pub fn runtime(&self) -> Option<&Arc<dyn Runtime>> {
713        self.runtime.as_ref()
714    }
715
716    /// The composed UI surface, admin actions included, for tooling
717    /// (`fz`, the control plane). `GET /__surface` serves the same
718    /// document, public subset unless the admin bearer is presented.
719    #[must_use]
720    pub fn surface(&self) -> &SurfaceDocument {
721        &self.surface.document
722    }
723
724    /// Assembles the full router: each module nested under `/v1/<name>`,
725    /// the one `/.well-known` router (if any) nested at the root,
726    /// `GET /__health`, `GET /__ready`, `GET /__surface`, the UI renderer
727    /// at `/ui` when one is mounted, and the shared middleware
728    /// (request-id/Scope, CORS allowlist, 64 KiB body limit, `/v1/*`
729    /// security headers, no-store headers for any request carrying a
730    /// `token` query parameter — issue #135). Nothing but `/.well-known`,
731    /// `/ui` and the `/__*` probes is ever mounted at the root.
732    /// What `/__health` answers from.
733    ///
734    /// Extracted from [`Harness::router`] only because that function is
735    /// at the line limit; it is one expression and it belongs to the
736    /// router's assembly.
737    fn health_state(&self, ports: &Ports, sidecars: Vec<SidecarMount>) -> HealthState {
738        HealthState {
739            // The deployment's environment, not the compiled default:
740            // `cratefield-waitlist` answered `"env":"development"` while
741            // serving production, which is the same lie #143 removed from
742            // the gates and left here.
743            venture: self.venture_as_deployed(ports.config.as_ref()),
744            modules: self.modules.clone(),
745            harness_build: ports.config.get("HARNESS_BUILD").filter(|b| !b.is_empty()),
746            mailer_configured: ports.mailer.is_some(),
747            captcha_configured: ports.captcha.is_some(),
748            sidecars,
749            dispatcher: ports.dispatcher.clone(),
750            clock: ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
751            probe_cache: new_probe_cache(),
752        }
753    }
754
755    pub fn router(&self, ports: Ports) -> Router {
756        // The environment the *deployment* declares, not only the one
757        // compiled in (issue #143). The limiter and signer legs read the
758        // resolved ports, not the runtime's advertisement (issues #437,
759        // #478) — captured before `ports` is destructured below.
760        let env = deployed_env(self.venture.env, ports.config.as_ref());
761        let (limiter, signer) = (ports.rate_limiter.is_some(), ports.signer.is_some());
762        let readiness = self.production_readiness_now(env, ports.config.as_ref(), limiter, signer);
763        // The problem `type` base of *this* venture, resolved once per
764        // router (issue #557): the venture's override, else
765        // `<public_url>/problems/`, else `about:blank`.
766        let problem_type_base: Arc<str> = self.venture.problem_type_base().into();
767        let mut api = self.nest_modules(&ports);
768        // The gateway signer is this deployment's half of the sidecar
769        // trust boundary (issue #131): absent secret, absent capability.
770        let gateway = gateway_signer(
771            ports.config.as_ref(),
772            ports.clock.clone().unwrap_or_else(|| Arc::new(SystemClock)),
773        );
774        let mounted = self.sidecar_mounts(&ports);
775        api = Self::nest_sidecars(api, &mounted, &ports, gateway.as_ref());
776        let gateway_state = Self::gateway_guard_state(&ports, gateway.as_ref());
777        let mounts_for_health = mounted.clone();
778        // Captured before `ports` is destructured below.
779        let tenant_layer = Arc::new(TenantLayer {
780            routing: ports.tenants.clone(),
781            fallback: ports.db.clone(),
782        });
783        // The harness is the floor of the admin plane's abuse control
784        // (issue #437): a module that forgets to limit its own `/admin`
785        // routes still gets the budget, namespaced under `admin:` so a
786        // guess at the admin bearer cannot share — or exhaust — a public
787        // route's. Modules that limit as well are double-limited, which
788        // is the floor's cost and not a bug. Applied here — after the
789        // module *and* sidecar nests — because a layer only wraps the
790        // routes present when it is applied: applied inside either nest,
791        // the other's `/admin/*` paths would sit outside the floor.
792        // Innermost of the four, so it spends budget only on requests
793        // that survived tenant resolution.
794        // Resolution last before the module routes, and on them only
795        // (TENANT-ROUTING.md §3): `/__health`, `/__ready`, `/.well-known`
796        // and `/ui` have no tenant, and resolving there would 404 every
797        // liveness probe in production.
798        //
799        // The streaming request layer (issue #585) is applied first, so it
800        // is the innermost of all: the `413` a declared over-ceiling body
801        // earns still passes back out through security headers, tenant
802        // resolution and the request-id/CORS layers, and a streaming route
803        // gets the same abuse-floor budget every other route does.
804        let streaming = StreamingState::build(&self.modules);
805        let api = api.layer(from_fn_with_state(streaming, streaming_request_layer));
806        let api = with_abuse_floor(api, ports.rate_limiter.clone(), tenant_layer);
807
808        let surface_source = self.merged_surface(mounted, &ports, gateway.clone(), env);
809
810        let ui = self.ui_router(&api, &ports, &surface_source);
811
812        let health_state = self.health_state(&ports, mounts_for_health);
813        // Readiness must know what the composition asked for but this
814        // bundle cannot answer, and it asks before `ports` is destructured
815        // below.
816        let missing_ports = missing_module_ports(&self.modules, &ports);
817        let Ports {
818            config,
819            db,
820            auth: _,
821            mailer: _,
822            captcha: _,
823            rate_limiter: _,
824            signer: _,
825            kv: _,
826            blob: _,
827            push: _,
828            payments: _,
829            tracker: _,
830            realtime: _,
831            text_model: _,
832            classifier: _,
833            vector_index: _,
834            embedder: _,
835            custom_hostnames: _,
836            http: _,
837            clock,
838            id_gen,
839            defer,
840            dispatcher: _,
841            tenants: _,
842        } = ports;
843
844        let scope_state = ScopeState {
845            defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
846            id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
847            // The identity stamp names a module only where there is exactly
848            // one to name: the sidecar role (issue #61).
849            module: (self.modules.len() == 1).then(|| self.modules[0].name()),
850        };
851        let ready_state = ReadyState {
852            db,
853            clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
854            missing_ports,
855        };
856
857        let surface_state = SurfaceState {
858            config,
859            source: surface_source,
860        };
861
862        let root = self.root_router(api, ui, gateway, health_state, ready_state, surface_state);
863
864        // Innermost layer: the gate runs after `Scope`, so refusals carry
865        // the request id, and it wraps every route — nested `/v1/*`, `/ui`
866        // and `/__surface` alike (issue #131). Tower order: the layer
867        // applied first is the innermost, so this line must come before
868        // `scope_layer` in the chain.
869        // Innermost of all: a deployment that declares production and
870        // cannot satisfy its own declared abuse controls refuses the
871        // guarded routes rather than serving them unprotected (#143).
872        root.layer(from_fn_with_state(
873            Arc::new(readiness),
874            production_readiness_guard,
875        ))
876        .layer(from_fn_with_state(gateway_state, gateway_guard))
877        .layer(from_fn_with_state(scope_state, scope_layer))
878        .layer(axum::middleware::from_fn(token_response_layer))
879        .layer(cors_layer(&self.venture.cors_origins))
880        // Outermost of all (issue #557): every problem body is named under
881        // *this* venture's base, whichever layer produced it — a module
882        // handler, the tenant or readiness gates, a 429, a rejection from
883        // an extractor with no state at all. `Problem::into_response`
884        // renders context-free (`about:blank`) and stores itself in the
885        // response extensions; this layer re-renders the body and nothing
886        // else. Applied last, so it wraps the module nests, the sidecars,
887        // `/.well-known`, `/ui` and the fallback alike.
888        .layer(from_fn_with_state(problem_type_base, problem_type_layer))
889    }
890}
891
892/// The `/v1/<name>` module a path addresses, plus its remaining segments.
893/// `None` when the path is not under a module mount. A query or fragment is
894/// cut at the first `?`/`#` (a caller passing a whole URL must not have the
895/// module misidentified from its tail) and empty segments from a leading,
896/// trailing or doubled slash are collapsed — [`Harness::max_body_bytes`]'
897/// lenient normalisation, which tolerates a path a caller typed by hand.
898/// Streaming classification uses the stricter [`exact_module_and_segments`].
899fn module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
900    let path = match path.split_once(['?', '#']) {
901        Some((path, _)) => path,
902        None => path,
903    };
904    let mut segments = path.split('/').filter(|segment| !segment.is_empty());
905    if segments.next() != Some("v1") {
906        return None;
907    }
908    let name = segments.next()?;
909    Some((name, segments.collect()))
910}
911
912/// The exact-segment counterpart of [`module_and_segments`], for classifying
913/// **streaming** routes. It collapses nothing: a trailing or doubled slash
914/// is a segment of its own, so `/upload/` and `//upload` match no route and
915/// are left to axum's own routing (a 404) — streaming must never claim a
916/// request axum would not serve. `path` must be a root-relative URL path (a
917/// leading `/`), which is what every request's `uri().path()` is.
918fn exact_module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
919    let path = match path.split_once(['?', '#']) {
920        Some((path, _)) => path,
921        None => path,
922    };
923    let mut segments = path.split('/');
924    // The split before the leading slash is an empty segment.
925    if segments.next() != Some("") || segments.next() != Some("v1") {
926        return None;
927    }
928    let name = segments.next()?;
929    if name.is_empty() {
930        return None;
931    }
932    Some((name, segments.collect()))
933}
934
935/// Whether `method` + `segments` matches one of `routes`, returning the
936/// route's ceiling.
937fn match_streaming(routes: &[StreamRoute], method: &Method, segments: &[&str]) -> Option<usize> {
938    routes.iter().find_map(|route| {
939        (route.method == *method && pattern_matches(route.path, segments))
940            .then_some(route.max_bytes)
941    })
942}
943
944/// Matches one route pattern's segments against a request path's: a literal
945/// segment compares equal, `{param}` matches exactly one segment, and
946/// `{*rest}` matches the remaining one-or-more and must be last.
947fn pattern_matches(pattern: &str, segments: &[&str]) -> bool {
948    let mut pattern_segments = pattern.split('/').filter(|segment| !segment.is_empty());
949    let mut rest = segments.iter();
950    loop {
951        match pattern_segments.next() {
952            Some("{*rest}") => {
953                return pattern_segments.next().is_none() && rest.next().is_some();
954            }
955            Some(param) if param.starts_with('{') && param.ends_with('}') => {
956                if rest.next().is_none() {
957                    return false;
958                }
959            }
960            Some(literal) => {
961                if rest.next() != Some(&literal) {
962                    return false;
963                }
964            }
965            None => return rest.next().is_none(),
966        }
967    }
968}
969
970/// Every module's streaming routes, snapshotted once per router so the
971/// request layer can match without touching the module list.
972#[derive(Clone)]
973struct StreamingState {
974    modules: Arc<Vec<(&'static str, &'static [StreamRoute])>>,
975}
976
977impl StreamingState {
978    fn build(modules: &[Arc<dyn Module>]) -> Self {
979        // Only modules that declared a route are kept, so a deployment with
980        // no streaming routes short-circuits with an empty vec.
981        let routes = modules
982            .iter()
983            .filter(|module| !module.streaming_routes().is_empty())
984            .map(|module| (module.name(), module.streaming_routes()))
985            .collect();
986        Self {
987            modules: Arc::new(routes),
988        }
989    }
990
991    fn max_bytes(&self, path: &str, method: &Method) -> Option<usize> {
992        let (name, segments) = exact_module_and_segments(path)?;
993        let (_, routes) = self.modules.iter().find(|(module, _)| *module == name)?;
994        match_streaming(routes, method, &segments)
995    }
996}
997
998/// Turns the body of a declared streaming route into a [`RequestStream`]
999/// (issue #585): the innermost layer of the API stack, so request id, CORS,
1000/// security headers, the abuse floor and the tenant gate all still apply,
1001/// and the `413` a declared over-ceiling `content-length` earns is decorated
1002/// by request id and CORS like any other problem.
1003///
1004/// A runtime that already broke the body into a stream — `runtime-cloudflare`
1005/// — pre-inserts its own [`RequestStream`] and hands an empty body; this sees
1006/// the extension and leaves the request untouched. Every route whose
1007/// `(method, path)` names no declared route passes straight through with its
1008/// buffered body and `DefaultBodyLimit` intact.
1009async fn streaming_request_layer(
1010    State(state): State<StreamingState>,
1011    mut request: axum::extract::Request,
1012    next: Next,
1013) -> Response {
1014    if state.modules.is_empty() {
1015        return next.run(request).await;
1016    }
1017    let Some(max_bytes) = state.max_bytes(request.uri().path(), request.method()) else {
1018        return next.run(request).await;
1019    };
1020    if request.extensions().get::<RequestStream>().is_some() {
1021        return next.run(request).await;
1022    }
1023    // Refused before any byte is read: a declared length over the route's
1024    // ceiling is the same `413 request-too-large` a buffered route answers.
1025    if let Some(declared) = crate::ports::declared_content_length(request.headers())
1026        && declared > max_bytes
1027    {
1028        let problem = Problem::request_too_large();
1029        let problem = match request.extensions().get::<Scope>() {
1030            Some(scope) => problem.instance(&scope.request_id),
1031            None => problem,
1032        };
1033        return problem.into_response();
1034    }
1035    let body = std::mem::replace(request.body_mut(), axum::body::Body::empty());
1036    request
1037        .extensions_mut()
1038        .insert(RequestStream::new(body.into_data_stream(), max_bytes));
1039    next.run(request).await
1040}
1041
1042/// The outer four layers every request passes, innermost last in source
1043/// order: the abuse floor (issue #437), tenant resolution, security
1044/// headers, body limit. Extracted so `Harness::router` stays under
1045/// clippy's line cap.
1046fn with_abuse_floor(
1047    api: Router,
1048    rate_limiter: Option<Arc<dyn RateLimiter>>,
1049    tenant_layer: Arc<TenantLayer>,
1050) -> Router {
1051    api.layer(from_fn_with_state(rate_limiter, admin_rate_limit_layer))
1052        .layer(axum::middleware::from_fn_with_state(
1053            tenant_layer,
1054            resolve_tenant_layer,
1055        ))
1056        .layer(axum::middleware::from_fn(security_headers_layer))
1057        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
1058}
1059
1060/// The outermost response layer (issue #557): re-renders a problem body
1061/// under the serving venture's base. Status, headers and everything a
1062/// module added ride through untouched — only the body is replaced, and
1063/// the now-stale `content-length` with it.
1064async fn problem_type_layer(
1065    State(base): State<Arc<str>>,
1066    request: axum::extract::Request,
1067    next: Next,
1068) -> Response {
1069    let mut response = next.run(request).await;
1070    let Some(problem) = response.extensions().get::<Problem>().cloned() else {
1071        return response;
1072    };
1073    let body = problem.body(&base).to_string();
1074    *response.body_mut() = axum::body::Body::from(body);
1075    response.headers_mut().remove(header::CONTENT_LENGTH);
1076    // Idempotence: drop the carriage once named. A harness router nested
1077    // in a larger service — whose own outermost layer may carry a *different*
1078    // venture's base — must not be re-rendered a second time.
1079    response.extensions_mut().remove::<Problem>();
1080    response
1081}
1082
1083/// What one `/__health` sidecar fan-out produced: when it ran (epoch ms)
1084/// and one entry per mount.
1085type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
1086
1087/// The fresh cache every router starts with. The shared mutex lives in one
1088/// place so the scoped allow for it has one justification to point at.
1089#[allow(clippy::disallowed_types)]
1090fn new_probe_cache() -> Arc<std::sync::Mutex<ProbeCache>> {
1091    Arc::new(std::sync::Mutex::new(None))
1092}
1093
1094#[derive(Clone)]
1095struct HealthState {
1096    venture: Arc<Venture>,
1097    modules: Vec<Arc<dyn Module>>,
1098    /// Git sha injected as a var by the deploy workflow (issue #14).
1099    harness_build: Option<String>,
1100    mailer_configured: bool,
1101    captcha_configured: bool,
1102    sidecars: Vec<SidecarMount>,
1103    dispatcher: Option<Arc<dyn Dispatcher>>,
1104    clock: Arc<dyn Clock>,
1105    /// The last sidecar probe, kept for one [`SIDECAR_PROBE_TTL_SECS`]
1106    /// window. The stamp on each forwarded response is the real contract
1107    /// check (issue #61); this probe exists only so `/__health` can show
1108    /// each sidecar's state, and caching it keeps a polling dashboard from
1109    /// turning every health check into a fan-out of subrequests.
1110    /// Deployment-scoped, not request state: it outlives no request, so
1111    /// ADR 0007's ban on ambient request state does not apply.
1112    #[allow(clippy::disallowed_types)]
1113    probe_cache: Arc<std::sync::Mutex<ProbeCache>>,
1114}
1115
1116/// How long a `/__health` sidecar listing stays fresh (issue #61). Short on
1117/// purpose: the point of the listing is to notice a redeploy.
1118const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1119
1120/// Per-sidecar probe budget, same generosity the readiness probe grants
1121/// the database.
1122const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1123
1124async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1125    let modules: Vec<serde_json::Value> = state
1126        .modules
1127        .iter()
1128        .map(|module| {
1129            json!({
1130                "name": module.name(),
1131                "version": module.version(),
1132                "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1133                "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1134                "tables": module.tables(),
1135                "emits": module.emits(),
1136            })
1137        })
1138        .collect();
1139    let sidecars = probe_sidecars(&state).await;
1140    Json(json!({
1141        "venture": state.venture.name,
1142        "env": state.venture.env.as_str(),
1143        "harness_api": HARNESS_API,
1144        "harness_build": state.harness_build,
1145        // Port presence: the Mailer/Captcha traits carry no probe, so a
1146        // NotConfigured adapter still reports its port as configured.
1147        "mailer": if state.mailer_configured { "configured" } else { "not_configured" },
1148        "captcha": if state.captcha_configured { "configured" } else { "absent" },
1149        "modules": modules,
1150        "sidecars": sidecars,
1151    }))
1152}
1153
1154/// One mount, probed. `probe` is the verdict (`ok`, `mismatch`,
1155/// `unreachable`); `contract`, `module` and `version` are what the sidecar
1156/// said about itself, `null` when it said nothing. A sidecar that answers
1157/// a wrong contract still reports here — the operator reading `/__health`
1158/// needs both numbers, and the per-request refusal is a separate story.
1159async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1160    if state.sidecars.is_empty() {
1161        return Vec::new();
1162    }
1163    let now_ms =
1164        i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1165    if let Some((at_ms, entries)) = state.probe_cache.lock().unwrap().as_ref()
1166        && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1167    {
1168        return entries.clone();
1169    }
1170
1171    let mut entries = Vec::new();
1172    for mount in &state.sidecars {
1173        entries.push(probe_one_sidecar(state, mount).await);
1174    }
1175    *state.probe_cache.lock().unwrap() = Some((now_ms, entries.clone()));
1176    entries
1177}
1178
1179async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1180    let mut missing = json!({
1181        "name": mount.name,
1182        "binding": mount.binding,
1183        "probe": "unreachable",
1184        "contract": serde_json::Value::Null,
1185        "module": serde_json::Value::Null,
1186        "version": serde_json::Value::Null,
1187    });
1188    let Some(dispatcher) = state.dispatcher.as_ref() else {
1189        return missing;
1190    };
1191    let request = http::Request::builder()
1192        .method(http::Method::GET)
1193        .uri("/__health")
1194        .body(Bytes::new());
1195    let Ok(request) = request else {
1196        return missing;
1197    };
1198    // `/__health` stays open on the sidecar side of the gateway (probes
1199    // must probe), so no stamp is needed to reach it. The future is owned
1200    // because the clock's timeout demands `'static`; the binding outlives
1201    // the call only as a clone.
1202    let dispatcher = Arc::clone(dispatcher);
1203    let binding = mount.binding.clone();
1204    let probe = async move { dispatcher.dispatch(&binding, request).await };
1205    let Some(Ok(response)) =
1206        crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1207    else {
1208        return missing;
1209    };
1210    let contract = response
1211        .headers()
1212        .get(crate::sidecar::X_HARNESS_API)
1213        .and_then(|v| v.to_str().ok())
1214        .and_then(|v| v.parse::<u32>().ok());
1215    let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1216    let module = response
1217        .headers()
1218        .get(crate::sidecar::X_HARNESS_MODULE)
1219        .and_then(|v| v.to_str().ok())
1220        .map(str::to_owned)
1221        .or_else(|| first_module_string(body.as_ref(), "name"));
1222    let version = first_module_string(body.as_ref(), "version");
1223    let tables = declared_tables(body.as_ref());
1224    let contract = contract.or_else(|| {
1225        body.as_ref()
1226            .and_then(|b| b.get("harness_api"))
1227            .and_then(serde_json::Value::as_u64)
1228            .and_then(|api| u32::try_from(api).ok())
1229    });
1230    missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1231    missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1232    missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1233    missing["tables"] = json!(tables);
1234    match contract {
1235        Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1236        Some(_) => missing["probe"] = json!("ok"),
1237        None => missing["probe"] = json!("unreachable"),
1238    }
1239    // A sidecar owns tables in the *same* database as its host, and the
1240    // build-time duplicate check (`check_tables`) cannot see it: it walks
1241    // `harness.modules()`, which a mount is not in. So the clash is caught
1242    // here, the first time the host hears what the sidecar claims, and it
1243    // is reported rather than inferred - `CREATE TABLE IF NOT EXISTS`
1244    // would otherwise have the two modules quietly sharing one table
1245    // (issue #66).
1246    //
1247    // Only a *positive* observation sets this. An unreachable sidecar
1248    // declares nothing, and "declared nothing" must never read as "clean":
1249    // that is the difference between a check and a coin flip.
1250    let clashes = collisions_with(&tables, state, &mount.name);
1251    if !clashes.is_empty() {
1252        let detail = format!(
1253            "sidecar `{}` claims table(s) {} already claimed in this deployment",
1254            mount.name,
1255            clashes.join(", ")
1256        );
1257        tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1258        crate::logging::forward_internal_error(&detail);
1259        missing["probe"] = json!("table-collision");
1260        missing["table_collisions"] = json!(clashes);
1261    }
1262    missing
1263}
1264
1265/// Every table named by `modules[*].tables` in a sidecar's `/__health`
1266/// body. Absent or malformed reads as an empty set, never as "no clash" -
1267/// see the caller.
1268fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1269    let Some(modules) = body
1270        .and_then(|b| b.get("modules"))
1271        .and_then(|m| m.as_array())
1272    else {
1273        return Vec::new();
1274    };
1275    let mut tables: Vec<String> = modules
1276        .iter()
1277        .filter_map(|module| module.get("tables")?.as_array())
1278        .flatten()
1279        .filter_map(|table| table.as_str().map(str::to_owned))
1280        .collect();
1281    tables.sort_unstable();
1282    tables.dedup();
1283    tables
1284}
1285
1286/// The tables `claimed` shares with a compiled-in module, reported with the
1287/// owner so the operator knows which side to rename.
1288fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1289    let mut clashes: Vec<String> = Vec::new();
1290    for module in &state.modules {
1291        if module.name() == mount {
1292            continue;
1293        }
1294        for table in module.tables() {
1295            if claimed.iter().any(|claimed| claimed == table)
1296                && !clashes.iter().any(|seen| seen == table)
1297            {
1298                clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1299            }
1300        }
1301    }
1302    clashes
1303}
1304
1305/// A sidecar's `/__health` body names its module and version at
1306/// `modules[0]`; a stamp header takes precedence, so this is the fallback.
1307fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1308    body?
1309        .get("modules")?
1310        .get(0)?
1311        .get(key)?
1312        .as_str()
1313        .map(str::to_owned)
1314}
1315
1316/// What the resolution layer needs: the deployment's tenant plane, and
1317/// the handle to fall back to when it has none.
1318struct TenantLayer {
1319    routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1320    /// The single database of a deployment without a registry. `None`
1321    /// when no database port is configured at all, which is a legitimate
1322    /// harness (a venture of pure-HTTP modules) — the tenant still
1323    /// resolves, and a handler that asks for a `TenantConn` is the thing
1324    /// that fails.
1325    fallback: Option<Arc<dyn Database>>,
1326}
1327
1328/// Resolves the request's tenant and puts it, with its database handle,
1329/// into the extensions for the `TenantConn` extractor (issue #32).
1330///
1331/// Refusals carry the request id because this runs inside `scope_layer` —
1332/// a refusal from outside the harness carries no `instance`, which is why
1333/// #143's readiness guard was pushed inside rather than left at the edge.
1334async fn resolve_tenant_layer(
1335    State(layer): State<Arc<TenantLayer>>,
1336    scope: Scope,
1337    mut request: axum::extract::Request,
1338    next: Next,
1339) -> Response {
1340    let host = request
1341        .headers()
1342        .get(header::HOST)
1343        .and_then(|value| value.to_str().ok())
1344        .unwrap_or_default()
1345        .to_owned();
1346
1347    // No tenant plane: the *no registry* deployment. Every host is the one
1348    // venture (TENANT-ROUTING.md §6), so a module is written once against
1349    // the stricter shape and the path most ventures run in production is
1350    // not the one without the isolation.
1351    let resolution = match layer.routing.as_ref() {
1352        Some(routing) => routing.resolve(&host),
1353        None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1354            &crate::tenant::ImplicitTenant,
1355            &host,
1356        ),
1357    };
1358
1359    let tenant = match resolution.admit() {
1360        Ok(tenant) => tenant,
1361        Err(problem) => {
1362            return Problem::new(problem)
1363                .instance(&scope.request_id)
1364                .into_response();
1365        }
1366    };
1367
1368    let db = match layer.routing.as_ref() {
1369        Some(routing) => match routing.database(&tenant).await {
1370            Ok(db) => Some(db),
1371            Err(err) => {
1372                // The registry is the only thing that knows a DSN, and
1373                // this error carries the tenant id and nothing else.
1374                tracing::error!(tenant = %tenant.id(), "{err}");
1375                crate::logging::forward_internal_error(&err.to_string());
1376                return Problem::new(&SLUGS.tenant_degraded)
1377                    .instance(&scope.request_id)
1378                    .into_response();
1379            }
1380        },
1381        None => layer.fallback.clone(),
1382    };
1383
1384    if let Some(db) = db {
1385        // Which shape resolved this tenant, carried through to the
1386        // handler: a membership rule must be able to tell a registry
1387        // deployment from a no-registry one, because the two disagree
1388        // about who a verified caller is (#385).
1389        let tenancy = if layer.routing.is_some() {
1390            crate::tenant::Tenancy::FromRegistry
1391        } else {
1392            crate::tenant::Tenancy::Sole
1393        };
1394        request
1395            .extensions_mut()
1396            .insert(crate::tenant_conn::ResolvedTenant {
1397                tenant,
1398                tenancy,
1399                db,
1400            });
1401    }
1402    next.run(request).await
1403}
1404
1405#[derive(Clone)]
1406struct ReadyState {
1407    db: Option<Arc<dyn Database>>,
1408    clock: Arc<dyn Clock>,
1409    /// Ports some composed module requires but this bundle did not resolve
1410    /// — see [`missing_module_ports`]. A deployment whose module cannot
1411    /// run is not ready, whatever the database says.
1412    missing_ports: Vec<Port>,
1413}
1414
1415/// The probed ports ([`Port::VectorIndex`], [`Port::Embedder`]) some
1416/// composed module requires but this bundle does not provide (issue #561):
1417/// the DB has its own leg of the probe, and an unwired port here means a
1418/// module that would answer every request with `NotConfigured`.
1419fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1420    const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1421    PROBED
1422        .iter()
1423        .copied()
1424        .filter(|port| {
1425            modules
1426                .iter()
1427                .any(|module| module.requires().contains(port))
1428                && !ports.has(*port)
1429        })
1430        .collect()
1431}
1432
1433/// `GET /__ready`: every port a module requires must be resolved, then
1434/// `SELECT 1` through the `Database` port; 503 problem on failure
1435/// (architecture section 6).
1436async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1437    if !state.missing_ports.is_empty() {
1438        let detail = state
1439            .missing_ports
1440            .iter()
1441            .map(|port| format!("{} port is not configured", port.name()))
1442            .collect::<Vec<_>>()
1443            .join("; ");
1444        return Problem::not_ready(detail).into_response();
1445    }
1446    let Some(db) = state.db else {
1447        return Problem::not_ready("database port is not configured").into_response();
1448    };
1449    let stmt = Statement::new("SELECT 1");
1450    let query = async move { db.query(&stmt).await };
1451    match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1452        Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1453        Some(Err(err)) => {
1454            error!(error = %err, "readiness probe query failed");
1455            crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1456            Problem::not_ready("database query failed").into_response()
1457        }
1458        None => Problem::not_ready("database did not answer within 2 s").into_response(),
1459    }
1460}
1461
1462#[derive(Clone)]
1463struct SurfaceState {
1464    config: Arc<dyn Config>,
1465    source: Arc<dyn SurfaceSource>,
1466}
1467
1468/// The build-time surface plus whatever the mounted sidecars answer
1469/// (issue #76). Sidecar surfaces are fetched on every call: a sidecar's
1470/// own `/__surface` is prerendered, the service binding runs on the same
1471/// thread (ADR 0009), and a cache here would hide a redeploy. Only the
1472/// public part of a sidecar merges: its admin routes take its own token,
1473/// which this host does not hold.
1474struct MergedSurface {
1475    base: Arc<SurfaceVariants>,
1476    mounts: Vec<SidecarMount>,
1477    dispatcher: Option<Arc<dyn Dispatcher>>,
1478    /// Mints the gateway stamp for `/__surface` fetches: the host is not
1479    /// exempt from the boundary it enforces on others (issue #131).
1480    gateway: Option<Arc<HmacSigner>>,
1481    env: VentureEnv,
1482    captcha_configured: bool,
1483}
1484
1485impl MergedSurface {
1486    /// What each mounted sidecar contributes, in mount order.
1487    async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1488        let mut extra = Vec::new();
1489        let Some(dispatcher) = &self.dispatcher else {
1490            return extra;
1491        };
1492        for mount in &self.mounts {
1493            if !dispatcher.has(&mount.binding) {
1494                continue;
1495            }
1496            let mut builder = axum::http::Request::builder()
1497                .method(axum::http::Method::GET)
1498                .uri("/__surface")
1499                .header(header::ACCEPT, "application/json");
1500            if let Some(signer) = &self.gateway {
1501                // A surface fetch is not an admin request: the plain purpose.
1502                builder =
1503                    builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1504            }
1505            let request = builder
1506                .body(bytes::Bytes::new())
1507                .expect("static request builds");
1508            let answer = match dispatcher.dispatch(&mount.binding, request).await {
1509                Ok(response) if response.status().is_success() => response,
1510                Ok(response) => {
1511                    tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1512                    continue;
1513                }
1514                Err(err) => {
1515                    tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1516                    continue;
1517                }
1518            };
1519            // Cap before parsing: an unbounded document turns a merge into
1520            // a memory attack on this Worker (issue #131).
1521            if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1522                tracing::warn!(
1523                    module = mount.name,
1524                    "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1525                );
1526                continue;
1527            }
1528            match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1529                Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1530                    Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1531                        // A production host with no Captcha port cannot
1532                        // render the widget a merged action demands, so the
1533                        // honest surface drops it (issue #131).
1534                        if self.env == VentureEnv::Production
1535                            && !self.captcha_configured
1536                            && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1537                        {
1538                            tracing::warn!(
1539                                module = entry.name,
1540                                "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1541                            );
1542                        }
1543                        (!entry.surface.is_empty()).then_some(entry)
1544                    })),
1545                    Err(problems) => {
1546                        for problem in problems {
1547                            tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1548                        }
1549                    }
1550                },
1551                Err(err) => {
1552                    tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1553                }
1554            }
1555        }
1556        extra
1557    }
1558}
1559
1560#[async_trait::async_trait]
1561impl SurfaceSource for MergedSurface {
1562    async fn current(&self) -> Arc<SurfaceDocument> {
1563        if self.mounts.is_empty() {
1564            return Arc::clone(&self.base.document);
1565        }
1566        let mut document = (*self.base.document).clone();
1567        document.modules.extend(self.sidecar_modules().await);
1568        Arc::new(document)
1569    }
1570
1571    fn built(&self) -> Arc<SurfaceDocument> {
1572        Arc::clone(&self.base.document)
1573    }
1574
1575    fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1576        Some(if admin {
1577            &self.base.full
1578        } else {
1579            &self.base.public
1580        })
1581    }
1582}
1583
1584/// `GET /__surface` (ADR 0010): the composed surface, public subset by
1585/// default, admin actions included when `Authorization: Bearer
1586/// <ADMIN_TOKEN>` is valid. A wrong or stale bearer is not an error here,
1587/// it just gets the public document: this route exists to be read by
1588/// renderers and tooling, and a `403` would leak whether admin is on.
1589/// Strong `ETag` per variant; `If-None-Match` answers `304`.
1590///
1591/// Both variants share this URL, so caching is split per variant (issue
1592/// #130): the public document stays `no-cache` (shared caches may store
1593/// it but must revalidate — nothing in it is a secret), while the
1594/// authenticated document is `private, no-store` on the `200` *and* on
1595/// the `304`, because a `304` refreshes what a cache already holds.
1596/// `Vary: Authorization` alone is not enough: an intermediary that
1597/// ignores `Vary` could otherwise store the admin variant and expose it
1598/// to an unauthenticated caller.
1599async fn surface_handler(
1600    State(state): State<SurfaceState>,
1601    headers: HeaderMap,
1602) -> impl IntoResponse {
1603    let admin = require_admin(&*state.config, &headers).is_ok();
1604    // With no sidecar the source hands back the build-time Arc, and the
1605    // prerendered variants are reused; with sidecars the merged document
1606    // is rendered per request (a hash, microseconds).
1607    let current = state.source.current().await;
1608    let built = state.source.built();
1609    let prerendered = Arc::ptr_eq(&current, &built).then(|| state.source.rendered(admin));
1610    let fresh;
1611    let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1612        rendered
1613    } else {
1614        fresh = if admin {
1615            RenderedSurface::render(&current)
1616        } else {
1617            RenderedSurface::render(&current.public())
1618        };
1619        &fresh
1620    };
1621    let matches = headers
1622        .get(header::IF_NONE_MATCH)
1623        .and_then(|value| value.to_str().ok())
1624        .is_some_and(|value| {
1625            value
1626                .split(',')
1627                .map(str::trim)
1628                .any(|tag| tag == "*" || tag == rendered.etag)
1629        });
1630    let mut response = if matches {
1631        StatusCode::NOT_MODIFIED.into_response()
1632    } else {
1633        (
1634            [(header::CONTENT_TYPE, "application/json")],
1635            rendered.json.clone(),
1636        )
1637            .into_response()
1638    };
1639    let response_headers = response.headers_mut();
1640    response_headers.insert(
1641        header::ETAG,
1642        header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1643    );
1644    response_headers.insert(
1645        header::CACHE_CONTROL,
1646        if admin {
1647            header::HeaderValue::from_static("private, no-store")
1648        } else {
1649            header::HeaderValue::from_static("no-cache")
1650        },
1651    );
1652    response_headers.insert(
1653        header::VARY,
1654        header::HeaderValue::from_static("Authorization"),
1655    );
1656    response
1657}
1658
1659/// Builder: `.venture(..)`, `.module(..)`, `.runtime(..)`, `.template(..)`,
1660/// then `.build()`.
1661#[derive(Default)]
1662pub struct HarnessBuilder {
1663    venture: Option<Venture>,
1664    modules: Vec<Arc<dyn Module>>,
1665    provides: Vec<Port>,
1666    runtime: Option<Arc<dyn Runtime>>,
1667    module_templates: Vec<(String, Box<dyn Template>)>,
1668    overrides: Vec<(String, Box<dyn Template>)>,
1669    ui: Option<Arc<dyn UiMount>>,
1670}
1671
1672impl HarnessBuilder {
1673    #[must_use]
1674    pub fn venture(mut self, venture: Venture) -> Self {
1675        self.venture = Some(venture);
1676        self
1677    }
1678
1679    /// Adds a module. Composition is compile-time: the wasm binary contains
1680    /// exactly the modules listed here (ADR 0003).
1681    #[must_use]
1682    pub fn module(mut self, module: impl Module) -> Self {
1683        self.modules.push(Arc::new(module));
1684        self
1685    }
1686
1687    /// Adds an already-shared module (`cratefield-testing` keeps handles to
1688    /// apply migrations and run conformance).
1689    #[must_use]
1690    pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1691        self.modules.push(module);
1692        self
1693    }
1694
1695    /// Declares the runtime: its `provides()` set drives build-time
1696    /// checking of every module's `requires()`. The runtime is kept on the
1697    /// built harness for tooling (`fz doctor`, scheduled fan-out).
1698    #[must_use]
1699    pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1700        let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1701        self.provides = runtime.provides();
1702        self.runtime = Some(runtime);
1703        self
1704    }
1705
1706    /// Registers module default templates (`<module>/<template>` ids).
1707    /// Call before overrides; see `template.rs` for the convention.
1708    #[must_use]
1709    pub fn templates(
1710        mut self,
1711        templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1712    ) -> Self {
1713        self.module_templates.extend(templates);
1714        self
1715    }
1716
1717    /// Mounts a UI renderer at `/ui` (ADR 0010): `cratefield_ui::Ui`. Off
1718    /// unless called, so a venture without a UI serves nothing there.
1719    #[must_use]
1720    pub fn ui(mut self, ui: impl UiMount) -> Self {
1721        self.ui = Some(Arc::new(ui));
1722        self
1723    }
1724
1725    /// Venture template override. Wins over any module default with the
1726    /// same id; the id's module part must name a registered module.
1727    #[must_use]
1728    pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1729        self.overrides.push((id.into(), template));
1730        self
1731    }
1732
1733    /// Validates everything, collecting **all** problems before failing
1734    /// (issue #2).
1735    ///
1736    /// # Errors
1737    ///
1738    /// `Err` whose `Display` lists every problem: invalid venture, unknown
1739    /// or duplicated port declarations, duplicate module names, tables or
1740    /// `/.well-known` routers, `harness_api` mismatches, invalid UI
1741    /// surfaces, unprovided required ports, and template ids naming
1742    /// unregistered modules.
1743    pub fn build(self) -> Result<Harness, ConfigError> {
1744        let mut errors = ConfigError::default();
1745
1746        let venture = if let Some(venture) = self.venture {
1747            venture.validate(&mut errors);
1748            venture
1749        } else {
1750            errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1751            Venture::new("invalid", "invalid.invalid")
1752        };
1753
1754        let mut names: HashMap<&'static str, usize> = HashMap::new();
1755        let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1756
1757        for module in &self.modules {
1758            if module.harness_api() != HARNESS_API {
1759                errors.push(harness_api_mismatch(module.as_ref()));
1760            }
1761
1762            let name = module.name();
1763            if name.is_empty() || !is_module_name(name) {
1764                errors.push(format!(
1765                    "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1766                ));
1767            }
1768            if names.insert(name, 1).is_some() {
1769                errors.push(format!("duplicate module name `{name}`"));
1770            }
1771
1772            for port in module.requires().iter().chain(module.optional()) {
1773                if !Port::ALL.contains(port) {
1774                    errors.push(format!(
1775                        "module `{name}` declares unknown port {}",
1776                        port.name()
1777                    ));
1778                }
1779            }
1780            for port in module.requires() {
1781                if module.optional().contains(port) {
1782                    errors.push(format!(
1783                        "module `{name}` lists port {} in both requires() and optional()",
1784                        port.name()
1785                    ));
1786                }
1787            }
1788
1789            module.surface().validate(name, &mut errors);
1790
1791            check_tables(module.as_ref(), &mut tables, &mut errors);
1792        }
1793
1794        let well_known = collect_well_known(&self.modules, &mut errors);
1795
1796        for module in &self.modules {
1797            for port in module.requires() {
1798                if !self.provides.contains(port) {
1799                    errors.push(format!(
1800                        "module `{}` requires port {} which the runtime does not provide",
1801                        module.name(),
1802                        port.name()
1803                    ));
1804                }
1805            }
1806            warn_undeclared_ports(module.as_ref(), &self.provides);
1807        }
1808
1809        let ordered = resolve_dependency_order(&self.modules, &mut errors);
1810
1811        append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1812
1813        check_template_ids(
1814            self.overrides.iter().chain(self.module_templates.iter()),
1815            &names,
1816            &mut errors,
1817        );
1818
1819        let surface = Arc::new(SurfaceVariants::compose(
1820            &venture,
1821            &self.modules,
1822            self.ui.as_ref(),
1823        ));
1824        if let Some(ui) = &self.ui {
1825            ui.validate(&surface.document, &mut errors);
1826        }
1827
1828        errors.into_result()?;
1829
1830        let mut registry = TemplateRegistry::new();
1831        registry.register_all(self.module_templates);
1832        registry.register_all(self.overrides);
1833
1834        let mut events = EventBus::new();
1835        for module in &self.modules {
1836            for (name, handler) in module.events() {
1837                events = events.on(name, handler);
1838            }
1839        }
1840
1841        Ok(Harness {
1842            venture: Arc::new(venture),
1843            unprotected_acceptance_recorded: AtomicBool::new(false),
1844            unlimited_acceptance_recorded: AtomicBool::new(false),
1845            // Dependency order, which is composition order until a module
1846            // declares something (RECONCILIATION.md §2).
1847            // Composed from the resolved order, so the catalog reads the way
1848            // the schema applies: a reader tracing an export down the list
1849            // meets a table before the tables that point at it.
1850            personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1851                ordered.iter().map(|m| (m.name(), m.personal_data())),
1852            )),
1853            modules: ordered,
1854            templates: Arc::new(registry),
1855            events,
1856            runtime: self.runtime,
1857            well_known,
1858            surface,
1859            ui: self.ui,
1860        })
1861    }
1862}
1863
1864/// A template id's module part must name a registered module.
1865/// Orders modules so that every module follows the ones it declares in
1866/// [`Module::depends_on`] (RECONCILIATION.md §2), reporting an unknown
1867/// dependency or a cycle as a build error rather than a boot one.
1868///
1869/// The sort is **stable**: modules with no dependency between them keep
1870/// composition order. With nothing declared anywhere — which is every
1871/// venture today — the result is the input, so adopting this changes no
1872/// ordering that already exists.
1873///
1874/// Returns composition order when the graph is unusable, so `build`
1875/// collects the rest of its errors instead of stopping at the first.
1876fn resolve_dependency_order(
1877    modules: &[Arc<dyn Module>],
1878    errors: &mut ConfigError,
1879) -> Vec<Arc<dyn Module>> {
1880    let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1881    let mut unknown = false;
1882    for module in modules {
1883        for needed in module.depends_on() {
1884            if !names.contains(needed) {
1885                unknown = true;
1886                errors.push(format!(
1887                    "module `{}` depends on `{needed}`, which this venture does not compose: \
1888                     add the module or drop the dependency",
1889                    module.name()
1890                ));
1891            }
1892        }
1893    }
1894    if unknown {
1895        return modules.to_vec();
1896    }
1897
1898    // Kahn's algorithm over composition order, which is what makes the
1899    // result stable: at each step the earliest-composed ready module is
1900    // taken, so an undeclared pair never reorders.
1901    let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
1902        .iter()
1903        .map(|module| Some(Arc::clone(module)))
1904        .collect();
1905    let mut placed: BTreeSet<&str> = BTreeSet::new();
1906    let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
1907
1908    while ordered.len() < modules.len() {
1909        let mut progressed = false;
1910        for slot in &mut remaining {
1911            let ready = slot.as_ref().is_some_and(|module| {
1912                module
1913                    .depends_on()
1914                    .iter()
1915                    .all(|needed| placed.contains(needed))
1916            });
1917            if ready {
1918                let module = slot.take().expect("checked above");
1919                placed.insert(module.name());
1920                ordered.push(module);
1921                progressed = true;
1922            }
1923        }
1924        if !progressed {
1925            // Everything left is waiting on something else left: a cycle.
1926            let stuck: Vec<&str> = remaining
1927                .iter()
1928                .filter_map(|slot| slot.as_ref().map(|module| module.name()))
1929                .collect();
1930            errors.push(format!(
1931                "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
1932                 that every module follows the ones it sits on top of",
1933                stuck.join(", ")
1934            ));
1935            return modules.to_vec();
1936        }
1937    }
1938    ordered
1939}
1940
1941fn check_template_ids<'a>(
1942    ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
1943    names: &HashMap<&'static str, usize>,
1944    errors: &mut ConfigError,
1945) {
1946    for (id, _) in ids {
1947        let Some(module_name) = id.split('/').next() else {
1948            continue;
1949        };
1950        if !names.contains_key(module_name) {
1951            errors.push(format!(
1952                "template `{id}` names module `{module_name}` which is not registered"
1953            ));
1954        }
1955    }
1956}
1957
1958fn is_module_name(name: &str) -> bool {
1959    name.split('-').all(|part| {
1960        !part.is_empty()
1961            && part
1962                .chars()
1963                .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
1964    })
1965}
1966
1967/// Refuses every guarded route when the deployment declares production
1968/// and cannot satisfy the abuse controls its own routes declare
1969/// (issue #143).
1970///
1971/// Fail *closed*, not fatal: a panicking Worker is an outage with no
1972/// diagnosis, so the probes and the UI stay up to say why while `/v1/*`
1973/// answers `503 not-production-ready`. `Harness::build` still refuses the
1974/// same composition outright when the venture declares its environment
1975/// honestly; this catches the deployment that did not.
1976async fn production_readiness_guard(
1977    State(problems): State<Arc<Vec<String>>>,
1978    request: axum::extract::Request,
1979    next: Next,
1980) -> Response {
1981    if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
1982        return next.run(request).await;
1983    }
1984    let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
1985    match request.extensions().get::<Scope>() {
1986        Some(scope) => problem.instance(&scope.request_id.clone()),
1987        None => problem,
1988    }
1989    .into_response()
1990}
1991
1992/// Limits every `/admin/*` route by client IP at the harness layer
1993/// (issue #437), whether or not the module limits its own. The keys are
1994/// prefixed `admin:` so a guess at the admin bearer draws from its own
1995/// budget, the way the auth crates namespace theirs (`auth-password:<key>`),
1996/// and no email key joins them — the caller is anonymous until it
1997/// authenticates. A limiter transport error denies: the admin token has no
1998/// captcha or cooldown behind it, so there is nothing to fail open into
1999/// (the sidecar forward reaches the same verdict for the same reason).
2000async fn admin_rate_limit_layer(
2001    State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2002    request: axum::extract::Request,
2003    next: Next,
2004) -> Response {
2005    if !crate::sidecar::is_admin_path(request.uri().path()) {
2006        return next.run(request).await;
2007    }
2008    let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2009        crate::rate_limit::client_ip(request.headers()).as_deref(),
2010        None,
2011    )
2012    .into_iter()
2013    .map(|key| format!("admin:{key}"))
2014    .collect();
2015    if let RateLimit::Denied { decision } =
2016        crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2017            .await
2018    {
2019        return rate_limited(&decision);
2020    }
2021    next.run(request).await
2022}
2023
2024/// Collects the modules' `/.well-known` routers (issue #46): at most one
2025/// module may provide one — `/.well-known` is a singleton discovery
2026/// namespace — and more is a build error naming every provider.
2027/// Production abuse controls are an initialization rule, not a doctor
2028/// suggestion (issue #133): a venture that boots in production cannot
2029/// rely on captcha or payment verification it does not actually have.
2030/// The overrides are deliberately `None` here — the operator's recorded
2031/// acceptances belong to the boot gate, which reads them from the
2032/// deployment config against the resolved ports. The webhook-secret leg
2033/// (`webhook_secret_readiness`, issue #533) is skipped for the same
2034/// reason: it reads the deployment config, which a build has none of.
2035fn append_production_readiness(
2036    venture: &Venture,
2037    modules: &[Arc<dyn Module>],
2038    runtime: Option<&Arc<dyn Runtime>>,
2039    errors: &mut ConfigError,
2040) {
2041    for error in crate::route_policy::production_readiness(
2042        venture.env,
2043        &crate::route_policy::WriteGuards::collect(modules),
2044        runtime,
2045        // Build time has no resolved ports, so these are the runtime's own
2046        // answers — provisionally optimistic for a binding that is named
2047        // but fails to resolve, or a secret that fails to parse. The boot
2048        // gate re-checks against the ports the runtime actually handed
2049        // over, and refuses there.
2050        crate::route_policy::rate_limiter_effective(runtime),
2051        crate::route_policy::signer_effective(runtime),
2052        None,
2053        None,
2054    ) {
2055        errors.push(error);
2056    }
2057}
2058
2059fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2060    let mut providers: Vec<&'static str> = Vec::new();
2061    let mut well_known = None;
2062    for module in modules {
2063        if let Some(router) = module.well_known() {
2064            providers.push(module.name());
2065            well_known = Some(router);
2066        }
2067    }
2068    if providers.len() > 1 {
2069        let listed = providers
2070            .iter()
2071            .map(|name| format!("`{name}`"))
2072            .collect::<Vec<_>>()
2073            .join(", ");
2074        errors.push(format!(
2075            "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2076        ));
2077    }
2078    well_known
2079}
2080
2081#[cfg(test)]
2082mod acceptance_recording {
2083    //! The once-per-deployment recording of an operator's escape-hatch
2084    //! acceptance (issue #143, #437). In-file because the guards are
2085    //! private `AtomicBool`s on `Harness`, and per-`Harness` rather than a
2086    //! process static so each test's harness carries its own record.
2087
2088    use super::*;
2089    use crate::Migrations;
2090    use crate::config::MapConfig;
2091
2092    /// A surface-less public writer — the conservative-fallback shape,
2093    /// which the captcha leg satisfies through the runtime below.
2094    struct PubWriter;
2095
2096    impl Module for PubWriter {
2097        fn name(&self) -> &'static str {
2098            "writer"
2099        }
2100        fn version(&self) -> &'static str {
2101            "0.0.0-test"
2102        }
2103        fn requires(&self) -> &'static [Port] {
2104            &[]
2105        }
2106        fn public_writes(&self) -> bool {
2107            true
2108        }
2109        fn migrations(&self) -> Migrations {
2110            Migrations::default()
2111        }
2112        fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2113            Ok(())
2114        }
2115        fn router(&self, _: ModuleContext) -> Router {
2116            Router::new()
2117        }
2118    }
2119
2120    /// Advertises every port, so the captcha leg is satisfied and each
2121    /// test below is about the limiter leg alone.
2122    struct AllPorts;
2123
2124    impl Runtime for AllPorts {
2125        fn provides(&self) -> Vec<Port> {
2126            Port::ALL.to_vec()
2127        }
2128    }
2129
2130    fn harness() -> Harness {
2131        Harness::builder()
2132            .venture(
2133                Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2134            )
2135            .module(PubWriter)
2136            .runtime(AllPorts)
2137            .build()
2138            .expect("a development venture builds")
2139    }
2140
2141    fn deployed(unlimited: Option<&str>) -> MapConfig {
2142        let pairs = [("ENV", "production")].into_iter().chain(
2143            unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2144        );
2145        MapConfig::from_pairs(pairs)
2146    }
2147
2148    #[test]
2149    fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2150        let harness = harness();
2151        let config = deployed(Some("issue #437: limiter binding pending"));
2152        assert!(
2153            harness
2154                .production_readiness_now(VentureEnv::Production, &config, false, true)
2155                .is_empty(),
2156            "the recorded acceptance serves"
2157        );
2158        assert!(
2159            harness
2160                .unlimited_acceptance_recorded
2161                .load(Ordering::Relaxed)
2162        );
2163
2164        // A later router build — on Workers there is one per request —
2165        // neither refuses nor re-records.
2166        assert!(
2167            harness
2168                .production_readiness_now(VentureEnv::Production, &config, false, true)
2169                .is_empty()
2170        );
2171    }
2172
2173    #[test]
2174    fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2175        let harness = harness();
2176        let config = deployed(Some("unused: the limiter resolved"));
2177        assert!(
2178            harness
2179                .production_readiness_now(VentureEnv::Production, &config, true, true)
2180                .is_empty()
2181        );
2182        assert!(
2183            !harness
2184                .unlimited_acceptance_recorded
2185                .load(Ordering::Relaxed),
2186            "a waiver that removed no refusal is not a recorded acceptance"
2187        );
2188    }
2189
2190    #[test]
2191    fn a_refusing_deployment_records_no_acceptance_at_all() {
2192        let harness = harness();
2193        // The captcha waiver does not reach the limiter leg, so the gate
2194        // refuses — and an acceptance that left the deployment refusing
2195        // recorded nothing, for either key.
2196        let config = MapConfig::from_pairs([
2197            ("ENV", "production"),
2198            (
2199                crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2200                "captcha pending",
2201            ),
2202        ]);
2203        assert_eq!(
2204            harness
2205                .production_readiness_now(VentureEnv::Production, &config, false, true)
2206                .len(),
2207            1
2208        );
2209        assert!(
2210            !harness
2211                .unprotected_acceptance_recorded
2212                .load(Ordering::Relaxed)
2213        );
2214        assert!(
2215            !harness
2216                .unlimited_acceptance_recorded
2217                .load(Ordering::Relaxed)
2218        );
2219    }
2220
2221    /// A webhook receiver that verifies its deliveries with the core HMAC
2222    /// scheme (issue #533): a signature writer that asks nothing of the
2223    /// Payments port — its secret key is the whole production gate.
2224    struct HmacWriter;
2225
2226    impl Module for HmacWriter {
2227        fn name(&self) -> &'static str {
2228            "webhooks"
2229        }
2230        fn version(&self) -> &'static str {
2231            "0.0.0-test"
2232        }
2233        fn requires(&self) -> &'static [Port] {
2234            &[]
2235        }
2236        fn public_writes(&self) -> bool {
2237            true
2238        }
2239        fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2240            crate::route_policy::RoutePolicy::Signature
2241        }
2242        fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2243            crate::route_policy::SignatureVerification::Hmac {
2244                secret: "WEBHOOK_SECRET",
2245            }
2246        }
2247        fn migrations(&self) -> Migrations {
2248            Migrations::default()
2249        }
2250        fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2251            Ok(())
2252        }
2253        fn router(&self, _: ModuleContext) -> Router {
2254            Router::new()
2255        }
2256    }
2257
2258    #[test]
2259    fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2260        let harness = Harness::builder()
2261            .venture(
2262                Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2263            )
2264            .module(HmacWriter)
2265            .runtime(AllPorts)
2266            .build()
2267            .expect("a development venture builds");
2268        let deployed = MapConfig::from_pairs([("ENV", "production")]);
2269        let problems =
2270            harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2271        assert_eq!(problems.len(), 1, "{problems:?}");
2272        assert!(
2273            problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2274            "{}",
2275            problems[0]
2276        );
2277        assert!(
2278            !problems[0].contains("Payments"),
2279            "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2280            problems[0]
2281        );
2282
2283        // The secret configured: the same venture is ready to serve.
2284        let ready = MapConfig::from_pairs([
2285            ("ENV", "production"),
2286            ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2287        ]);
2288        assert!(
2289            harness
2290                .production_readiness_now(VentureEnv::Production, &ready, true, true)
2291                .is_empty()
2292        );
2293    }
2294}