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