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, Mailer, Port, Ports, RateLimiter, Statement,
38    SystemClock, 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: ports.mailer.clone(),
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            url_safety: _,
912            http: _,
913            clock,
914            id_gen,
915            defer,
916            actors: _,
917            dispatcher: _,
918            tenants: _,
919        } = ports;
920
921        let scope_state = ScopeState {
922            defer: defer.unwrap_or_else(|| Arc::new(crate::ports::NoopDefer)),
923            id_gen: id_gen.unwrap_or_else(|| Arc::new(crate::ports::UlidIdGen)),
924            // The identity stamp names a module only where there is exactly
925            // one to name: the sidecar role (issue #61).
926            module: (self.modules.len() == 1).then(|| self.modules[0].name()),
927        };
928        let ready_state = ReadyState {
929            db,
930            clock: clock.unwrap_or_else(|| Arc::new(SystemClock)),
931            missing_ports,
932        };
933
934        let surface_state = SurfaceState {
935            config,
936            source: surface_source,
937        };
938
939        let root = self.root_router(api, ui, gateway, health_state, ready_state, surface_state);
940
941        // Innermost layer: the gate runs after `Scope`, so refusals carry
942        // the request id, and it wraps every route — nested `/v1/*`, `/ui`
943        // and `/__surface` alike (issue #131). Tower order: the layer
944        // applied first is the innermost, so this line must come before
945        // `scope_layer` in the chain.
946        // Innermost of all: a deployment that declares production and
947        // cannot satisfy its own declared abuse controls refuses the
948        // guarded routes rather than serving them unprotected (#143).
949        root.layer(from_fn_with_state(
950            Arc::new(readiness),
951            production_readiness_guard,
952        ))
953        .layer(from_fn_with_state(gateway_state, gateway_guard))
954        .layer(from_fn_with_state(scope_state, scope_layer))
955        .layer(axum::middleware::from_fn(token_response_layer))
956        .layer(cors_layer(&self.venture.cors_origins))
957        // Outermost of all (issue #557): every problem body is named under
958        // *this* venture's base, whichever layer produced it — a module
959        // handler, the tenant or readiness gates, a 429, a rejection from
960        // an extractor with no state at all. `Problem::into_response`
961        // renders context-free (`about:blank`) and stores itself in the
962        // response extensions; this layer re-renders the body and nothing
963        // else. Applied last, so it wraps the module nests, the sidecars,
964        // `/.well-known`, `/ui` and the fallback alike.
965        .layer(from_fn_with_state(problem_type_base, problem_type_layer))
966    }
967}
968
969/// The `/v1/<name>` module a path addresses, plus its remaining segments.
970/// `None` when the path is not under a module mount. A query or fragment is
971/// cut at the first `?`/`#` (a caller passing a whole URL must not have the
972/// module misidentified from its tail) and empty segments from a leading,
973/// trailing or doubled slash are collapsed — [`Harness::max_body_bytes`]'
974/// lenient normalisation, which tolerates a path a caller typed by hand.
975/// Streaming classification uses the stricter [`exact_module_and_segments`].
976fn module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
977    let path = match path.split_once(['?', '#']) {
978        Some((path, _)) => path,
979        None => path,
980    };
981    let mut segments = path.split('/').filter(|segment| !segment.is_empty());
982    if segments.next() != Some("v1") {
983        return None;
984    }
985    let name = segments.next()?;
986    Some((name, segments.collect()))
987}
988
989/// The exact-segment counterpart of [`module_and_segments`], for classifying
990/// **streaming** routes. It collapses nothing: a trailing or doubled slash
991/// is a segment of its own, so `/upload/` and `//upload` match no route and
992/// are left to axum's own routing (a 404) — streaming must never claim a
993/// request axum would not serve. `path` must be a root-relative URL path (a
994/// leading `/`), which is what every request's `uri().path()` is.
995fn exact_module_and_segments(path: &str) -> Option<(&str, Vec<&str>)> {
996    let path = match path.split_once(['?', '#']) {
997        Some((path, _)) => path,
998        None => path,
999    };
1000    let mut segments = path.split('/');
1001    // The split before the leading slash is an empty segment.
1002    if segments.next() != Some("") || segments.next() != Some("v1") {
1003        return None;
1004    }
1005    let name = segments.next()?;
1006    if name.is_empty() {
1007        return None;
1008    }
1009    Some((name, segments.collect()))
1010}
1011
1012/// Whether `method` + `segments` matches one of `routes`, returning the
1013/// route's ceiling.
1014fn match_streaming(routes: &[StreamRoute], method: &Method, segments: &[&str]) -> Option<usize> {
1015    routes.iter().find_map(|route| {
1016        (route.method == *method && pattern_matches(route.path, segments))
1017            .then_some(route.max_bytes)
1018    })
1019}
1020
1021/// Matches one route pattern's segments against a request path's: a literal
1022/// segment compares equal, `{param}` matches exactly one segment, and
1023/// `{*rest}` matches the remaining one-or-more and must be last.
1024fn pattern_matches(pattern: &str, segments: &[&str]) -> bool {
1025    let mut pattern_segments = pattern.split('/').filter(|segment| !segment.is_empty());
1026    let mut rest = segments.iter();
1027    loop {
1028        match pattern_segments.next() {
1029            Some("{*rest}") => {
1030                return pattern_segments.next().is_none() && rest.next().is_some();
1031            }
1032            Some(param) if param.starts_with('{') && param.ends_with('}') => {
1033                if rest.next().is_none() {
1034                    return false;
1035                }
1036            }
1037            Some(literal) => {
1038                if rest.next() != Some(&literal) {
1039                    return false;
1040                }
1041            }
1042            None => return rest.next().is_none(),
1043        }
1044    }
1045}
1046
1047/// Every module's streaming routes, snapshotted once per router so the
1048/// request layer can match without touching the module list.
1049#[derive(Clone)]
1050struct StreamingState {
1051    modules: Arc<Vec<(&'static str, &'static [StreamRoute])>>,
1052}
1053
1054impl StreamingState {
1055    fn build(modules: &[Arc<dyn Module>]) -> Self {
1056        // Only modules that declared a route are kept, so a deployment with
1057        // no streaming routes short-circuits with an empty vec.
1058        let routes = modules
1059            .iter()
1060            .filter(|module| !module.streaming_routes().is_empty())
1061            .map(|module| (module.name(), module.streaming_routes()))
1062            .collect();
1063        Self {
1064            modules: Arc::new(routes),
1065        }
1066    }
1067
1068    fn max_bytes(&self, path: &str, method: &Method) -> Option<usize> {
1069        let (name, segments) = exact_module_and_segments(path)?;
1070        let (_, routes) = self.modules.iter().find(|(module, _)| *module == name)?;
1071        match_streaming(routes, method, &segments)
1072    }
1073}
1074
1075/// Turns the body of a declared streaming route into a [`RequestStream`]
1076/// (issue #585): the innermost layer of the API stack, so request id, CORS,
1077/// security headers, the abuse floor and the tenant gate all still apply,
1078/// and the `413` a declared over-ceiling `content-length` earns is decorated
1079/// by request id and CORS like any other problem.
1080///
1081/// A runtime that already broke the body into a stream — `runtime-cloudflare`
1082/// — pre-inserts its own [`RequestStream`] and hands an empty body; this sees
1083/// the extension and leaves the request untouched. Every route whose
1084/// `(method, path)` names no declared route passes straight through with its
1085/// buffered body and `DefaultBodyLimit` intact.
1086async fn streaming_request_layer(
1087    State(state): State<StreamingState>,
1088    mut request: axum::extract::Request,
1089    next: Next,
1090) -> Response {
1091    if state.modules.is_empty() {
1092        return next.run(request).await;
1093    }
1094    let Some(max_bytes) = state.max_bytes(request.uri().path(), request.method()) else {
1095        return next.run(request).await;
1096    };
1097    if request.extensions().get::<RequestStream>().is_some() {
1098        return next.run(request).await;
1099    }
1100    // Refused before any byte is read: a declared length over the route's
1101    // ceiling is the same `413 request-too-large` a buffered route answers.
1102    if let Some(declared) = crate::ports::declared_content_length(request.headers())
1103        && declared > max_bytes
1104    {
1105        let problem = Problem::request_too_large();
1106        let problem = match request.extensions().get::<Scope>() {
1107            Some(scope) => problem.instance(&scope.request_id),
1108            None => problem,
1109        };
1110        return problem.into_response();
1111    }
1112    let body = std::mem::replace(request.body_mut(), axum::body::Body::empty());
1113    request
1114        .extensions_mut()
1115        .insert(RequestStream::new(body.into_data_stream(), max_bytes));
1116    next.run(request).await
1117}
1118
1119/// The outer four layers every request passes, innermost last in source
1120/// order: the abuse floor (issue #437), tenant resolution, security
1121/// headers, body limit. Extracted so `Harness::router` stays under
1122/// clippy's line cap.
1123fn with_abuse_floor(
1124    api: Router,
1125    rate_limiter: Option<Arc<dyn RateLimiter>>,
1126    tenant_layer: Arc<TenantLayer>,
1127) -> Router {
1128    api.layer(from_fn_with_state(rate_limiter, admin_rate_limit_layer))
1129        .layer(axum::middleware::from_fn_with_state(
1130            tenant_layer,
1131            resolve_tenant_layer,
1132        ))
1133        .layer(axum::middleware::from_fn(security_headers_layer))
1134        .layer(DefaultBodyLimit::max(MAX_BODY_BYTES))
1135}
1136
1137/// The outermost response layer (issue #557): re-renders a problem body
1138/// under the serving venture's base. Status, headers and everything a
1139/// module added ride through untouched — only the body is replaced, and
1140/// the now-stale `content-length` with it.
1141async fn problem_type_layer(
1142    State(base): State<Arc<str>>,
1143    request: axum::extract::Request,
1144    next: Next,
1145) -> Response {
1146    let mut response = next.run(request).await;
1147    let Some(problem) = response.extensions().get::<Problem>().cloned() else {
1148        return response;
1149    };
1150    let body = problem.body(&base).to_string();
1151    *response.body_mut() = axum::body::Body::from(body);
1152    response.headers_mut().remove(header::CONTENT_LENGTH);
1153    // Idempotence: drop the carriage once named. A harness router nested
1154    // in a larger service — whose own outermost layer may carry a *different*
1155    // venture's base — must not be re-rendered a second time.
1156    response.extensions_mut().remove::<Problem>();
1157    response
1158}
1159
1160/// What one `/__health` sidecar fan-out produced: when it ran (epoch ms)
1161/// and one entry per mount.
1162type ProbeCache = Option<(i64, Vec<serde_json::Value>)>;
1163
1164/// The fresh cache every router starts with. The shared lock lives in one
1165/// place so the single-flight justification has one home to point at.
1166fn new_probe_cache() -> Arc<futures_util::lock::Mutex<ProbeCache>> {
1167    Arc::new(futures_util::lock::Mutex::new(None))
1168}
1169
1170#[derive(Clone)]
1171struct HealthState {
1172    venture: Arc<Venture>,
1173    modules: Vec<Arc<dyn Module>>,
1174    /// Git sha injected as a var by the deploy workflow (issue #14).
1175    harness_build: Option<String>,
1176    /// The mailer itself, not a yes/no: `/__health` lists its providers so
1177    /// a venture running more than one can see them all (issue #793).
1178    mailer: Option<Arc<dyn Mailer>>,
1179    captcha_configured: bool,
1180    sidecars: Vec<SidecarMount>,
1181    dispatcher: Option<Arc<dyn Dispatcher>>,
1182    clock: Arc<dyn Clock>,
1183    /// The last sidecar probe, kept for one [`SIDECAR_PROBE_TTL_SECS`]
1184    /// window. The stamp on each forwarded response is the real contract
1185    /// check (issue #61); this probe exists only so `/__health` can show
1186    /// each sidecar's state, and caching it keeps a polling dashboard from
1187    /// turning every health check into a fan-out of subrequests.
1188    /// Deployment-scoped, not request state: it outlives no request, so
1189    /// ADR 0007's ban on ambient request state does not apply. The lock is
1190    /// held across the probe so a burst of concurrent health checks runs one
1191    /// fan-out, not one each — the `Realtime` room table has the same shape
1192    /// for the same reason.
1193    probe_cache: Arc<futures_util::lock::Mutex<ProbeCache>>,
1194}
1195
1196/// How long a `/__health` sidecar listing stays fresh (issue #61). Short on
1197/// purpose: the point of the listing is to notice a redeploy.
1198const SIDECAR_PROBE_TTL_SECS: i64 = 30;
1199
1200/// Per-sidecar probe budget, same generosity the readiness probe grants
1201/// the database.
1202const SIDECAR_PROBE_TIMEOUT: Duration = Duration::from_secs(2);
1203
1204async fn health_handler(State(state): State<HealthState>) -> impl IntoResponse {
1205    let modules: Vec<serde_json::Value> = state
1206        .modules
1207        .iter()
1208        .map(|module| {
1209            json!({
1210                "name": module.name(),
1211                "version": module.version(),
1212                "requires": module.requires().iter().map(Port::name).collect::<Vec<_>>(),
1213                "optional": module.optional().iter().map(Port::name).collect::<Vec<_>>(),
1214                "tables": module.tables(),
1215                "emits": module.emits(),
1216            })
1217        })
1218        .collect();
1219    let sidecars = probe_sidecars(&state).await;
1220    // The per-provider listing (issue #793): names and verdicts only. An
1221    // adapter reports its own configured/unknown, and a secret never gets
1222    // this far — `MailProvider` has no field that could carry one.
1223    let mailers = state
1224        .mailer
1225        .as_ref()
1226        .map(|mailer| mailer.providers())
1227        .unwrap_or_default();
1228    Json(json!({
1229        "venture": state.venture.name,
1230        "env": state.venture.env.as_str(),
1231        "harness_api": HARNESS_API,
1232        "harness_build": state.harness_build,
1233        // Port presence, not capability: the Mailer/Captcha traits carry no
1234        // probe, so a NotConfigured adapter still reports its port as
1235        // configured. `mailers` above is the finer, per-adapter answer and
1236        // the two are meant to disagree — a wired Resend with no API key
1237        // reads `configured` here and `configured: false` there.
1238        "mailer": if state.mailer.is_some() { "configured" } else { "not_configured" },
1239        "captcha": if state.captcha_configured { "configured" } else { "absent" },
1240        "mailers": mailers,
1241        "modules": modules,
1242        "sidecars": sidecars,
1243    }))
1244}
1245
1246/// One mount, probed. `probe` is the verdict (`ok`, `mismatch`,
1247/// `unreachable`); `contract`, `module` and `version` are what the sidecar
1248/// said about itself, `null` when it said nothing. A sidecar that answers
1249/// a wrong contract still reports here — the operator reading `/__health`
1250/// needs both numbers, and the per-request refusal is a separate story.
1251///
1252/// One fan-out, shared: the cache is checked and filled under one lock held
1253/// across the probe, so health checks arriving together queue on it and then
1254/// read the entries the first one stored rather than each probing every
1255/// sidecar for itself. The stamp is taken after the fan-out, so the TTL
1256/// window starts when the answer was actually produced.
1257async fn probe_sidecars(state: &HealthState) -> Vec<serde_json::Value> {
1258    if state.sidecars.is_empty() {
1259        return Vec::new();
1260    }
1261    let mut cache = state.probe_cache.lock().await;
1262    let now_ms =
1263        i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1264    if let Some((at_ms, entries)) = cache.as_ref()
1265        && now_ms - at_ms < SIDECAR_PROBE_TTL_SECS * 1000
1266    {
1267        return entries.clone();
1268    }
1269
1270    let mut entries = Vec::new();
1271    for mount in &state.sidecars {
1272        entries.push(probe_one_sidecar(state, mount).await);
1273    }
1274    let stamped_ms =
1275        i64::try_from(state.clock.now().unix_timestamp_nanos() / 1_000_000).unwrap_or(i64::MAX);
1276    *cache = Some((stamped_ms, entries.clone()));
1277    entries
1278}
1279
1280async fn probe_one_sidecar(state: &HealthState, mount: &SidecarMount) -> serde_json::Value {
1281    let mut missing = json!({
1282        "name": mount.name,
1283        "binding": mount.binding,
1284        "probe": "unreachable",
1285        "contract": serde_json::Value::Null,
1286        "module": serde_json::Value::Null,
1287        "version": serde_json::Value::Null,
1288    });
1289    let Some(dispatcher) = state.dispatcher.as_ref() else {
1290        return missing;
1291    };
1292    let request = http::Request::builder()
1293        .method(http::Method::GET)
1294        .uri("/__health")
1295        .body(Bytes::new());
1296    let Ok(request) = request else {
1297        return missing;
1298    };
1299    // `/__health` stays open on the sidecar side of the gateway (probes
1300    // must probe), so no stamp is needed to reach it. The future is owned
1301    // because the clock's timeout demands `'static`; the binding outlives
1302    // the call only as a clone.
1303    let dispatcher = Arc::clone(dispatcher);
1304    let binding = mount.binding.clone();
1305    let probe = async move { dispatcher.dispatch(&binding, request).await };
1306    let Some(Ok(response)) =
1307        crate::ports::timeout(&*state.clock, probe, SIDECAR_PROBE_TIMEOUT).await
1308    else {
1309        return missing;
1310    };
1311    let contract = response
1312        .headers()
1313        .get(crate::sidecar::X_HARNESS_API)
1314        .and_then(|v| v.to_str().ok())
1315        .and_then(|v| v.parse::<u32>().ok());
1316    let body = serde_json::from_slice::<serde_json::Value>(response.body()).ok();
1317    let module = response
1318        .headers()
1319        .get(crate::sidecar::X_HARNESS_MODULE)
1320        .and_then(|v| v.to_str().ok())
1321        .map(str::to_owned)
1322        .or_else(|| first_module_string(body.as_ref(), "name"));
1323    let version = first_module_string(body.as_ref(), "version");
1324    let tables = declared_tables(body.as_ref());
1325    let contract = contract.or_else(|| {
1326        body.as_ref()
1327            .and_then(|b| b.get("harness_api"))
1328            .and_then(serde_json::Value::as_u64)
1329            .and_then(|api| u32::try_from(api).ok())
1330    });
1331    missing["contract"] = contract.map_or(serde_json::Value::Null, |api| json!(api));
1332    missing["module"] = module.map_or(serde_json::Value::Null, |name| json!(name));
1333    missing["version"] = version.map_or(serde_json::Value::Null, |version| json!(version));
1334    missing["tables"] = json!(tables);
1335    match contract {
1336        Some(api) if api != HARNESS_API => missing["probe"] = json!("mismatch"),
1337        Some(_) => missing["probe"] = json!("ok"),
1338        None => missing["probe"] = json!("unreachable"),
1339    }
1340    // A sidecar owns tables in the *same* database as its host, and the
1341    // build-time duplicate check (`check_tables`) cannot see it: it walks
1342    // `harness.modules()`, which a mount is not in. So the clash is caught
1343    // here, the first time the host hears what the sidecar claims, and it
1344    // is reported rather than inferred - `CREATE TABLE IF NOT EXISTS`
1345    // would otherwise have the two modules quietly sharing one table
1346    // (issue #66).
1347    //
1348    // Only a *positive* observation sets this. An unreachable sidecar
1349    // declares nothing, and "declared nothing" must never read as "clean":
1350    // that is the difference between a check and a coin flip.
1351    let clashes = collisions_with(&tables, state, &mount.name);
1352    if !clashes.is_empty() {
1353        let detail = format!(
1354            "sidecar `{}` claims table(s) {} already claimed in this deployment",
1355            mount.name,
1356            clashes.join(", ")
1357        );
1358        tracing::error!(sidecar = %mount.name, tables = %clashes.join(","), "{detail}");
1359        crate::logging::forward_internal_error(&detail);
1360        missing["probe"] = json!("table-collision");
1361        missing["table_collisions"] = json!(clashes);
1362    }
1363    missing
1364}
1365
1366/// Every table named by `modules[*].tables` in a sidecar's `/__health`
1367/// body. Absent or malformed reads as an empty set, never as "no clash" -
1368/// see the caller.
1369fn declared_tables(body: Option<&serde_json::Value>) -> Vec<String> {
1370    let Some(modules) = body
1371        .and_then(|b| b.get("modules"))
1372        .and_then(|m| m.as_array())
1373    else {
1374        return Vec::new();
1375    };
1376    let mut tables: Vec<String> = modules
1377        .iter()
1378        .filter_map(|module| module.get("tables")?.as_array())
1379        .flatten()
1380        .filter_map(|table| table.as_str().map(str::to_owned))
1381        .collect();
1382    tables.sort_unstable();
1383    tables.dedup();
1384    tables
1385}
1386
1387/// The tables `claimed` shares with a compiled-in module, reported with the
1388/// owner so the operator knows which side to rename.
1389fn collisions_with(claimed: &[String], state: &HealthState, mount: &str) -> Vec<String> {
1390    let mut clashes: Vec<String> = Vec::new();
1391    for module in &state.modules {
1392        if module.name() == mount {
1393            continue;
1394        }
1395        for table in module.tables() {
1396            if claimed.iter().any(|claimed| claimed == table)
1397                && !clashes.iter().any(|seen| seen == table)
1398            {
1399                clashes.push(format!("`{table}` (also module `{}`)", module.name()));
1400            }
1401        }
1402    }
1403    clashes
1404}
1405
1406/// A sidecar's `/__health` body names its module and version at
1407/// `modules[0]`; a stamp header takes precedence, so this is the fallback.
1408fn first_module_string(body: Option<&serde_json::Value>, key: &str) -> Option<String> {
1409    body?
1410        .get("modules")?
1411        .get(0)?
1412        .get(key)?
1413        .as_str()
1414        .map(str::to_owned)
1415}
1416
1417/// What the resolution layer needs: the deployment's tenant plane, and
1418/// the handle to fall back to when it has none.
1419struct TenantLayer {
1420    routing: Option<Arc<dyn crate::tenant::TenantRouting>>,
1421    /// The single database of a deployment without a registry. `None`
1422    /// when no database port is configured at all, which is a legitimate
1423    /// harness (a venture of pure-HTTP modules) — the tenant still
1424    /// resolves, and a handler that asks for a `TenantConn` is the thing
1425    /// that fails.
1426    fallback: Option<Arc<dyn Database>>,
1427}
1428
1429/// Resolves the request's tenant and puts it, with its database handle,
1430/// into the extensions for the `TenantConn` extractor (issue #32).
1431///
1432/// Refusals carry the request id because this runs inside `scope_layer` —
1433/// a refusal from outside the harness carries no `instance`, which is why
1434/// #143's readiness guard was pushed inside rather than left at the edge.
1435async fn resolve_tenant_layer(
1436    State(layer): State<Arc<TenantLayer>>,
1437    scope: Scope,
1438    mut request: axum::extract::Request,
1439    next: Next,
1440) -> Response {
1441    let host = request
1442        .headers()
1443        .get(header::HOST)
1444        .and_then(|value| value.to_str().ok())
1445        .unwrap_or_default()
1446        .to_owned();
1447
1448    // No tenant plane: the *no registry* deployment. Every host is the one
1449    // venture (TENANT-ROUTING.md §6), so a module is written once against
1450    // the stricter shape and the path most ventures run in production is
1451    // not the one without the isolation.
1452    let resolution = match layer.routing.as_ref() {
1453        Some(routing) => routing.resolve(&host),
1454        None => <crate::tenant::ImplicitTenant as crate::tenant::ResolveTenant>::resolve(
1455            &crate::tenant::ImplicitTenant,
1456            &host,
1457        ),
1458    };
1459
1460    let tenant = match resolution.admit() {
1461        Ok(tenant) => tenant,
1462        Err(problem) => {
1463            return Problem::new(problem)
1464                .instance(&scope.request_id)
1465                .into_response();
1466        }
1467    };
1468
1469    let db = match layer.routing.as_ref() {
1470        Some(routing) => match routing.database(&tenant).await {
1471            Ok(db) => Some(db),
1472            Err(err) => {
1473                // The registry is the only thing that knows a DSN, and
1474                // this error carries the tenant id and nothing else.
1475                tracing::error!(tenant = %tenant.id(), "{err}");
1476                crate::logging::forward_internal_error(&err.to_string());
1477                return Problem::new(&SLUGS.tenant_degraded)
1478                    .instance(&scope.request_id)
1479                    .into_response();
1480            }
1481        },
1482        None => layer.fallback.clone(),
1483    };
1484
1485    if let Some(db) = db {
1486        // Which shape resolved this tenant, carried through to the
1487        // handler: a membership rule must be able to tell a registry
1488        // deployment from a no-registry one, because the two disagree
1489        // about who a verified caller is (#385).
1490        let tenancy = if layer.routing.is_some() {
1491            crate::tenant::Tenancy::FromRegistry
1492        } else {
1493            crate::tenant::Tenancy::Sole
1494        };
1495        request
1496            .extensions_mut()
1497            .insert(crate::tenant_conn::ResolvedTenant {
1498                tenant,
1499                tenancy,
1500                db,
1501            });
1502    }
1503    next.run(request).await
1504}
1505
1506#[derive(Clone)]
1507struct ReadyState {
1508    db: Option<Arc<dyn Database>>,
1509    clock: Arc<dyn Clock>,
1510    /// Ports some composed module requires but this bundle did not resolve
1511    /// — see [`missing_module_ports`]. A deployment whose module cannot
1512    /// run is not ready, whatever the database says.
1513    missing_ports: Vec<Port>,
1514}
1515
1516/// The probed ports ([`Port::VectorIndex`], [`Port::Embedder`]) some
1517/// composed module requires but this bundle does not provide (issue #561):
1518/// the DB has its own leg of the probe, and an unwired port here means a
1519/// module that would answer every request with `NotConfigured`.
1520fn missing_module_ports(modules: &[Arc<dyn Module>], ports: &Ports) -> Vec<Port> {
1521    const PROBED: &[Port] = &[Port::VectorIndex, Port::Embedder];
1522    PROBED
1523        .iter()
1524        .copied()
1525        .filter(|port| {
1526            modules
1527                .iter()
1528                .any(|module| module.requires().contains(port))
1529                && !ports.has(*port)
1530        })
1531        .collect()
1532}
1533
1534/// `GET /__ready`: every port a module requires must be resolved, then
1535/// `SELECT 1` through the `Database` port; 503 problem on failure
1536/// (architecture section 6).
1537async fn ready_handler(State(state): State<ReadyState>) -> impl IntoResponse {
1538    if !state.missing_ports.is_empty() {
1539        let detail = state
1540            .missing_ports
1541            .iter()
1542            .map(|port| format!("{} port is not configured", port.name()))
1543            .collect::<Vec<_>>()
1544            .join("; ");
1545        return Problem::not_ready(detail).into_response();
1546    }
1547    let Some(db) = state.db else {
1548        return Problem::not_ready("database port is not configured").into_response();
1549    };
1550    let stmt = Statement::new("SELECT 1");
1551    let query = async move { db.query(&stmt).await };
1552    match crate::ports::timeout(&*state.clock, query, Duration::from_secs(2)).await {
1553        Some(Ok(_rows)) => Json(json!({ "ok": true })).into_response(),
1554        Some(Err(err)) => {
1555            error!(error = %err, "readiness probe query failed");
1556            crate::logging::forward_internal_error(&format!("readiness probe query failed: {err}"));
1557            Problem::not_ready("database query failed").into_response()
1558        }
1559        None => Problem::not_ready("database did not answer within 2 s").into_response(),
1560    }
1561}
1562
1563#[derive(Clone)]
1564struct SurfaceState {
1565    config: Arc<dyn Config>,
1566    source: Arc<dyn SurfaceSource>,
1567}
1568
1569/// The build-time surface plus whatever the mounted sidecars answer
1570/// (issue #76). Sidecar surfaces are fetched on every call: a sidecar's
1571/// own `/__surface` is prerendered, the service binding runs on the same
1572/// thread (ADR 0009), and a cache here would hide a redeploy. Only the
1573/// public part of a sidecar merges: its admin routes take its own token,
1574/// which this host does not hold.
1575struct MergedSurface {
1576    base: Arc<SurfaceVariants>,
1577    mounts: Vec<SidecarMount>,
1578    dispatcher: Option<Arc<dyn Dispatcher>>,
1579    /// Mints the gateway stamp for `/__surface` fetches: the host is not
1580    /// exempt from the boundary it enforces on others (issue #131).
1581    gateway: Option<Arc<HmacSigner>>,
1582    env: VentureEnv,
1583    captcha_configured: bool,
1584}
1585
1586impl MergedSurface {
1587    /// What each mounted sidecar contributes, in mount order.
1588    async fn sidecar_modules(&self) -> Vec<crate::surface::ModuleSurface> {
1589        let mut extra = Vec::new();
1590        let Some(dispatcher) = &self.dispatcher else {
1591            return extra;
1592        };
1593        for mount in &self.mounts {
1594            if !dispatcher.has(&mount.binding) {
1595                continue;
1596            }
1597            let mut builder = axum::http::Request::builder()
1598                .method(axum::http::Method::GET)
1599                .uri("/__surface")
1600                .header(header::ACCEPT, "application/json");
1601            if let Some(signer) = &self.gateway {
1602                // A surface fetch is not an admin request: the plain purpose.
1603                builder =
1604                    builder.header(X_HARNESS_GATEWAY, mint_gateway(signer, &mount.name, false));
1605            }
1606            let request = builder
1607                .body(bytes::Bytes::new())
1608                .expect("static request builds");
1609            let answer = match dispatcher.dispatch(&mount.binding, request).await {
1610                Ok(response) if response.status().is_success() => response,
1611                Ok(response) => {
1612                    tracing::warn!(module = mount.name, status = %response.status(), "sidecar surface not available");
1613                    continue;
1614                }
1615                Err(err) => {
1616                    tracing::warn!(module = mount.name, error = %err, "sidecar surface fetch failed");
1617                    continue;
1618                }
1619            };
1620            // Cap before parsing: an unbounded document turns a merge into
1621            // a memory attack on this Worker (issue #131).
1622            if answer.body().len() > MAX_SIDECAR_SURFACE_BYTES {
1623                tracing::warn!(
1624                    module = mount.name,
1625                    "sidecar surface exceeds {MAX_SIDECAR_SURFACE_BYTES} bytes and was not merged"
1626                );
1627                continue;
1628            }
1629            match serde_json::from_slice::<SurfaceDocument>(answer.body()) {
1630                Ok(document) => match sanitize_sidecar_document(&document, &mount.name) {
1631                    Ok(entries) => extra.extend(entries.into_iter().filter_map(|mut entry| {
1632                        // A production host with no Captcha port cannot
1633                        // render the widget a merged action demands, so the
1634                        // honest surface drops it (issue #131).
1635                        if self.env == VentureEnv::Production
1636                            && !self.captcha_configured
1637                            && strip_unguarded_captcha_actions(&mut entry.surface) > 0
1638                        {
1639                            tracing::warn!(
1640                                module = entry.name,
1641                                "merged sidecar actions requiring a captcha were dropped: no Captcha port here"
1642                            );
1643                        }
1644                        (!entry.surface.is_empty()).then_some(entry)
1645                    })),
1646                    Err(problems) => {
1647                        for problem in problems {
1648                            tracing::warn!(module = mount.name, error = problem, "sidecar surface rejected at merge");
1649                        }
1650                    }
1651                },
1652                Err(err) => {
1653                    tracing::warn!(module = mount.name, error = %err, "sidecar surface is not a surface document");
1654                }
1655            }
1656        }
1657        extra
1658    }
1659}
1660
1661#[async_trait::async_trait]
1662impl SurfaceSource for MergedSurface {
1663    async fn current(&self) -> Arc<SurfaceDocument> {
1664        if self.mounts.is_empty() {
1665            return Arc::clone(&self.base.document);
1666        }
1667        let mut document = (*self.base.document).clone();
1668        document.modules.extend(self.sidecar_modules().await);
1669        Arc::new(document)
1670    }
1671
1672    fn built(&self) -> Arc<SurfaceDocument> {
1673        Arc::clone(&self.base.document)
1674    }
1675
1676    fn rendered(&self, admin: bool) -> Option<&RenderedSurface> {
1677        Some(if admin {
1678            &self.base.full
1679        } else {
1680            &self.base.public
1681        })
1682    }
1683}
1684
1685/// `GET /__surface` (ADR 0010): the composed surface, public subset by
1686/// default, admin actions included when `Authorization: Bearer
1687/// <ADMIN_TOKEN>` is valid. A wrong or stale bearer is not an error here,
1688/// it just gets the public document: this route exists to be read by
1689/// renderers and tooling, and a `403` would leak whether admin is on.
1690/// Strong `ETag` per variant; `If-None-Match` answers `304`.
1691///
1692/// Both variants share this URL, so caching is split per variant (issue
1693/// #130): the public document stays `no-cache` (shared caches may store
1694/// it but must revalidate — nothing in it is a secret), while the
1695/// authenticated document is `private, no-store` on the `200` *and* on
1696/// the `304`, because a `304` refreshes what a cache already holds.
1697/// `Vary: Authorization` alone is not enough: an intermediary that
1698/// ignores `Vary` could otherwise store the admin variant and expose it
1699/// to an unauthenticated caller.
1700async fn surface_handler(
1701    State(state): State<SurfaceState>,
1702    headers: HeaderMap,
1703) -> impl IntoResponse {
1704    let admin = require_admin(&*state.config, &headers).is_ok();
1705    // With no sidecar the source hands back the build-time Arc, and the
1706    // prerendered variants are reused; with sidecars the merged document
1707    // is rendered per request (a hash, microseconds).
1708    let current = state.source.current().await;
1709    let built = state.source.built();
1710    let prerendered = Arc::ptr_eq(&current, &built).then(|| state.source.rendered(admin));
1711    let fresh;
1712    let rendered: &RenderedSurface = if let Some(rendered) = prerendered.flatten() {
1713        rendered
1714    } else {
1715        fresh = if admin {
1716            RenderedSurface::render(&current)
1717        } else {
1718            RenderedSurface::render(&current.public())
1719        };
1720        &fresh
1721    };
1722    let matches = headers
1723        .get(header::IF_NONE_MATCH)
1724        .and_then(|value| value.to_str().ok())
1725        .is_some_and(|value| {
1726            value
1727                .split(',')
1728                .map(str::trim)
1729                .any(|tag| tag == "*" || tag == rendered.etag)
1730        });
1731    let mut response = if matches {
1732        StatusCode::NOT_MODIFIED.into_response()
1733    } else {
1734        (
1735            [(header::CONTENT_TYPE, "application/json")],
1736            rendered.json.clone(),
1737        )
1738            .into_response()
1739    };
1740    let response_headers = response.headers_mut();
1741    response_headers.insert(
1742        header::ETAG,
1743        header::HeaderValue::from_str(&rendered.etag).expect("hex etag is a valid header"),
1744    );
1745    response_headers.insert(
1746        header::CACHE_CONTROL,
1747        if admin {
1748            header::HeaderValue::from_static("private, no-store")
1749        } else {
1750            header::HeaderValue::from_static("no-cache")
1751        },
1752    );
1753    response_headers.insert(
1754        header::VARY,
1755        header::HeaderValue::from_static("Authorization"),
1756    );
1757    response
1758}
1759
1760/// Builder: `.venture(..)`, `.module(..)`, `.runtime(..)`, `.template(..)`,
1761/// then `.build()`.
1762#[derive(Default)]
1763pub struct HarnessBuilder {
1764    venture: Option<Venture>,
1765    modules: Vec<Arc<dyn Module>>,
1766    provides: Vec<Port>,
1767    runtime: Option<Arc<dyn Runtime>>,
1768    module_templates: Vec<(String, Box<dyn Template>)>,
1769    overrides: Vec<(String, Box<dyn Template>)>,
1770    ui: Option<Arc<dyn UiMount>>,
1771    actor_handlers: Vec<(String, Arc<dyn ActorHandler>)>,
1772}
1773
1774impl HarnessBuilder {
1775    #[must_use]
1776    pub fn venture(mut self, venture: Venture) -> Self {
1777        self.venture = Some(venture);
1778        self
1779    }
1780
1781    /// Adds a module. Composition is compile-time: the wasm binary contains
1782    /// exactly the modules listed here (ADR 0003).
1783    #[must_use]
1784    pub fn module(mut self, module: impl Module) -> Self {
1785        self.modules.push(Arc::new(module));
1786        self
1787    }
1788
1789    /// Adds an already-shared module (`cratefield-testing` keeps handles to
1790    /// apply migrations and run conformance).
1791    #[must_use]
1792    pub fn module_arc(mut self, module: Arc<dyn Module>) -> Self {
1793        self.modules.push(module);
1794        self
1795    }
1796
1797    /// Declares the runtime: its `provides()` set drives build-time
1798    /// checking of every module's `requires()`. The runtime is kept on the
1799    /// built harness for tooling (`fz doctor`, scheduled fan-out).
1800    #[must_use]
1801    pub fn runtime(mut self, runtime: impl Runtime) -> Self {
1802        let runtime: Arc<dyn Runtime> = Arc::new(runtime);
1803        self.provides = runtime.provides();
1804        self.runtime = Some(runtime);
1805        self
1806    }
1807
1808    /// Registers module default templates (`<module>/<template>` ids).
1809    /// Call before overrides; see `template.rs` for the convention.
1810    #[must_use]
1811    pub fn templates(
1812        mut self,
1813        templates: impl IntoIterator<Item = (String, Box<dyn Template>)>,
1814    ) -> Self {
1815        self.module_templates.extend(templates);
1816        self
1817    }
1818
1819    /// Mounts a UI renderer at `/ui` (ADR 0010): `cratefield_ui::Ui`. Off
1820    /// unless called, so a venture without a UI serves nothing there.
1821    #[must_use]
1822    pub fn ui(mut self, ui: impl UiMount) -> Self {
1823        self.ui = Some(Arc::new(ui));
1824        self
1825    }
1826
1827    /// Venture template override. Wins over any module default with the
1828    /// same id; the id's module part must name a registered module.
1829    #[must_use]
1830    pub fn template(mut self, id: impl Into<String>, template: Box<dyn Template>) -> Self {
1831        self.overrides.push((id.into(), template));
1832        self
1833    }
1834
1835    /// Registers the handler for one actor kind (issue #583). A module that
1836    /// declares `kind` in `actor_kinds()` reaches it through the `actor`
1837    /// port; registering a kind twice, or declaring one with no handler, is a
1838    /// build error.
1839    #[must_use]
1840    pub fn actor(mut self, kind: &str, handler: Arc<dyn ActorHandler>) -> Self {
1841        self.actor_handlers.push((kind.to_owned(), handler));
1842        self
1843    }
1844
1845    /// Validates everything, collecting **all** problems before failing
1846    /// (issue #2).
1847    ///
1848    /// # Errors
1849    ///
1850    /// `Err` whose `Display` lists every problem: invalid venture, unknown
1851    /// or duplicated port declarations, duplicate module names, tables or
1852    /// `/.well-known` routers, `harness_api` mismatches, invalid UI
1853    /// surfaces, unprovided required ports, and template ids naming
1854    /// unregistered modules.
1855    pub fn build(self) -> Result<Harness, ConfigError> {
1856        let mut errors = ConfigError::default();
1857
1858        let venture = if let Some(venture) = self.venture {
1859            venture.validate(&mut errors);
1860            venture
1861        } else {
1862            errors.push("missing venture: call .venture(Venture::new(..)) before .build()");
1863            Venture::new("invalid", "invalid.invalid")
1864        };
1865
1866        let mut names: HashMap<&'static str, usize> = HashMap::new();
1867        let mut tables: HashMap<&'static str, &'static str> = HashMap::new();
1868
1869        for module in &self.modules {
1870            if module.harness_api() != HARNESS_API {
1871                errors.push(harness_api_mismatch(module.as_ref()));
1872            }
1873
1874            let name = module.name();
1875            if name.is_empty() || !is_module_name(name) {
1876                errors.push(format!(
1877                    "module name `{name}` must be kebab-case ([a-z0-9]+ separated by '-')"
1878                ));
1879            }
1880            if names.insert(name, 1).is_some() {
1881                errors.push(format!("duplicate module name `{name}`"));
1882            }
1883
1884            for port in module.requires().iter().chain(module.optional()) {
1885                if !Port::ALL.contains(port) {
1886                    errors.push(format!(
1887                        "module `{name}` declares unknown port {}",
1888                        port.name()
1889                    ));
1890                }
1891            }
1892            for port in module.requires() {
1893                if module.optional().contains(port) {
1894                    errors.push(format!(
1895                        "module `{name}` lists port {} in both requires() and optional()",
1896                        port.name()
1897                    ));
1898                }
1899            }
1900
1901            module.surface().validate(name, &mut errors);
1902
1903            check_tables(module.as_ref(), &mut tables, &mut errors);
1904        }
1905
1906        let actor_handlers = collect_actors(&self.actor_handlers, &self.modules, &mut errors);
1907
1908        let well_known = collect_well_known(&self.modules, &mut errors);
1909
1910        for module in &self.modules {
1911            for port in module.requires() {
1912                if !self.provides.contains(port) {
1913                    errors.push(format!(
1914                        "module `{}` requires port {} which the runtime does not provide",
1915                        module.name(),
1916                        port.name()
1917                    ));
1918                }
1919            }
1920            warn_undeclared_ports(module.as_ref(), &self.provides);
1921        }
1922
1923        let ordered = resolve_dependency_order(&self.modules, &mut errors);
1924
1925        append_production_readiness(&venture, &self.modules, self.runtime.as_ref(), &mut errors);
1926
1927        check_template_ids(
1928            self.overrides.iter().chain(self.module_templates.iter()),
1929            &names,
1930            &mut errors,
1931        );
1932
1933        let surface = Arc::new(SurfaceVariants::compose(
1934            &venture,
1935            &self.modules,
1936            self.ui.as_ref(),
1937        ));
1938        if let Some(ui) = &self.ui {
1939            ui.validate(&surface.document, &mut errors);
1940        }
1941
1942        errors.into_result()?;
1943
1944        let mut registry = TemplateRegistry::new();
1945        registry.register_all(self.module_templates);
1946        registry.register_all(self.overrides);
1947
1948        let mut events = EventBus::new();
1949        for module in &self.modules {
1950            for (name, handler) in module.events() {
1951                events = events.on(name, handler);
1952            }
1953        }
1954
1955        Ok(Harness {
1956            venture: Arc::new(venture),
1957            unprotected_acceptance_recorded: AtomicBool::new(false),
1958            unlimited_acceptance_recorded: AtomicBool::new(false),
1959            // Dependency order, which is composition order until a module
1960            // declares something (RECONCILIATION.md §2).
1961            // Composed from the resolved order, so the catalog reads the way
1962            // the schema applies: a reader tracing an export down the list
1963            // meets a table before the tables that point at it.
1964            personal_data: Arc::new(crate::PersonalDataCatalog::compose(
1965                ordered.iter().map(|m| (m.name(), m.personal_data())),
1966            )),
1967            modules: ordered,
1968            templates: Arc::new(registry),
1969            events,
1970            runtime: self.runtime,
1971            well_known,
1972            surface,
1973            ui: self.ui,
1974            actor_handlers,
1975        })
1976    }
1977}
1978
1979/// A template id's module part must name a registered module.
1980/// Orders modules so that every module follows the ones it declares in
1981/// [`Module::depends_on`] (RECONCILIATION.md §2), reporting an unknown
1982/// dependency or a cycle as a build error rather than a boot one.
1983///
1984/// The sort is **stable**: modules with no dependency between them keep
1985/// composition order. With nothing declared anywhere — which is every
1986/// venture today — the result is the input, so adopting this changes no
1987/// ordering that already exists.
1988///
1989/// Returns composition order when the graph is unusable, so `build`
1990/// collects the rest of its errors instead of stopping at the first.
1991fn resolve_dependency_order(
1992    modules: &[Arc<dyn Module>],
1993    errors: &mut ConfigError,
1994) -> Vec<Arc<dyn Module>> {
1995    let names: BTreeSet<&str> = modules.iter().map(|module| module.name()).collect();
1996    let mut unknown = false;
1997    for module in modules {
1998        for needed in module.depends_on() {
1999            if !names.contains(needed) {
2000                unknown = true;
2001                errors.push(format!(
2002                    "module `{}` depends on `{needed}`, which this venture does not compose: \
2003                     add the module or drop the dependency",
2004                    module.name()
2005                ));
2006            }
2007        }
2008    }
2009    if unknown {
2010        return modules.to_vec();
2011    }
2012
2013    // Kahn's algorithm over composition order, which is what makes the
2014    // result stable: at each step the earliest-composed ready module is
2015    // taken, so an undeclared pair never reorders.
2016    let mut remaining: Vec<Option<Arc<dyn Module>>> = modules
2017        .iter()
2018        .map(|module| Some(Arc::clone(module)))
2019        .collect();
2020    let mut placed: BTreeSet<&str> = BTreeSet::new();
2021    let mut ordered: Vec<Arc<dyn Module>> = Vec::with_capacity(modules.len());
2022
2023    while ordered.len() < modules.len() {
2024        let mut progressed = false;
2025        for slot in &mut remaining {
2026            let ready = slot.as_ref().is_some_and(|module| {
2027                module
2028                    .depends_on()
2029                    .iter()
2030                    .all(|needed| placed.contains(needed))
2031            });
2032            if ready {
2033                let module = slot.take().expect("checked above");
2034                placed.insert(module.name());
2035                ordered.push(module);
2036                progressed = true;
2037            }
2038        }
2039        if !progressed {
2040            // Everything left is waiting on something else left: a cycle.
2041            let stuck: Vec<&str> = remaining
2042                .iter()
2043                .filter_map(|slot| slot.as_ref().map(|module| module.name()))
2044                .collect();
2045            errors.push(format!(
2046                "modules [{}] depend on each other in a cycle: migrations cannot be ordered so \
2047                 that every module follows the ones it sits on top of",
2048                stuck.join(", ")
2049            ));
2050            return modules.to_vec();
2051        }
2052    }
2053    ordered
2054}
2055
2056fn check_template_ids<'a>(
2057    ids: impl Iterator<Item = &'a (String, Box<dyn Template>)>,
2058    names: &HashMap<&'static str, usize>,
2059    errors: &mut ConfigError,
2060) {
2061    for (id, _) in ids {
2062        let Some(module_name) = id.split('/').next() else {
2063            continue;
2064        };
2065        if !names.contains_key(module_name) {
2066            errors.push(format!(
2067                "template `{id}` names module `{module_name}` which is not registered"
2068            ));
2069        }
2070    }
2071}
2072
2073fn is_module_name(name: &str) -> bool {
2074    name.split('-').all(|part| {
2075        !part.is_empty()
2076            && part
2077                .chars()
2078                .all(|c| c.is_ascii_lowercase() || c.is_ascii_digit())
2079    })
2080}
2081
2082/// Refuses every guarded route when the deployment declares production
2083/// and cannot satisfy the abuse controls its own routes declare
2084/// (issue #143).
2085///
2086/// Fail *closed*, not fatal: a panicking Worker is an outage with no
2087/// diagnosis, so the probes and the UI stay up to say why while `/v1/*`
2088/// answers `503 not-production-ready`. `Harness::build` still refuses the
2089/// same composition outright when the venture declares its environment
2090/// honestly; this catches the deployment that did not.
2091async fn production_readiness_guard(
2092    State(problems): State<Arc<Vec<String>>>,
2093    request: axum::extract::Request,
2094    next: Next,
2095) -> Response {
2096    if problems.is_empty() || !request.uri().path().starts_with("/v1/") {
2097        return next.run(request).await;
2098    }
2099    let problem = Problem::new(&SLUGS.not_production_ready).with_detail(problems.join("; "));
2100    match request.extensions().get::<Scope>() {
2101        Some(scope) => problem.instance(&scope.request_id.clone()),
2102        None => problem,
2103    }
2104    .into_response()
2105}
2106
2107/// Limits every `/admin/*` route by client IP at the harness layer
2108/// (issue #437), whether or not the module limits its own. The keys are
2109/// prefixed `admin:` so a guess at the admin bearer draws from its own
2110/// budget, the way the auth crates namespace theirs (`auth-password:<key>`),
2111/// and no email key joins them — the caller is anonymous until it
2112/// authenticates. A limiter transport error denies: the admin token has no
2113/// captcha or cooldown behind it, so there is nothing to fail open into
2114/// (the sidecar forward reaches the same verdict for the same reason).
2115async fn admin_rate_limit_layer(
2116    State(limiter): State<Option<Arc<dyn RateLimiter>>>,
2117    request: axum::extract::Request,
2118    next: Next,
2119) -> Response {
2120    if !crate::sidecar::is_admin_path(request.uri().path()) {
2121        return next.run(request).await;
2122    }
2123    let keys: Vec<String> = crate::rate_limit::rate_limit_keys(
2124        crate::rate_limit::client_ip(request.headers()).as_deref(),
2125        None,
2126    )
2127    .into_iter()
2128    .map(|key| format!("admin:{key}"))
2129    .collect();
2130    if let RateLimit::Denied { decision } =
2131        crate::rate_limit::check_rate_limit(limiter.as_ref(), &keys, RateLimitFailure::FailClosed)
2132            .await
2133    {
2134        return rate_limited(&decision);
2135    }
2136    next.run(request).await
2137}
2138
2139/// Collects the modules' `/.well-known` routers (issue #46): at most one
2140/// module may provide one — `/.well-known` is a singleton discovery
2141/// namespace — and more is a build error naming every provider.
2142/// Production abuse controls are an initialization rule, not a doctor
2143/// suggestion (issue #133): a venture that boots in production cannot
2144/// rely on captcha or payment verification it does not actually have.
2145/// The overrides are deliberately `None` here — the operator's recorded
2146/// acceptances belong to the boot gate, which reads them from the
2147/// deployment config against the resolved ports. The webhook-secret leg
2148/// (`webhook_secret_readiness`, issue #533) is skipped for the same
2149/// reason: it reads the deployment config, which a build has none of.
2150fn append_production_readiness(
2151    venture: &Venture,
2152    modules: &[Arc<dyn Module>],
2153    runtime: Option<&Arc<dyn Runtime>>,
2154    errors: &mut ConfigError,
2155) {
2156    for error in crate::route_policy::production_readiness(
2157        venture.env,
2158        &crate::route_policy::WriteGuards::collect(modules),
2159        runtime,
2160        // Build time has no resolved ports, so these are the runtime's own
2161        // answers — provisionally optimistic for a binding that is named
2162        // but fails to resolve, or a secret that fails to parse. The boot
2163        // gate re-checks against the ports the runtime actually handed
2164        // over, and refuses there.
2165        crate::route_policy::rate_limiter_effective(runtime),
2166        crate::route_policy::signer_effective(runtime),
2167        None,
2168        None,
2169    ) {
2170        errors.push(error);
2171    }
2172}
2173
2174fn collect_well_known(modules: &[Arc<dyn Module>], errors: &mut ConfigError) -> Option<Router> {
2175    let mut providers: Vec<&'static str> = Vec::new();
2176    let mut well_known = None;
2177    for module in modules {
2178        if let Some(router) = module.well_known() {
2179            providers.push(module.name());
2180            well_known = Some(router);
2181        }
2182    }
2183    if providers.len() > 1 {
2184        let listed = providers
2185            .iter()
2186            .map(|name| format!("`{name}`"))
2187            .collect::<Vec<_>>()
2188            .join(", ");
2189        errors.push(format!(
2190            "modules {listed} all provide a well-known router; at most one module may occupy /.well-known"
2191        ));
2192    }
2193    well_known
2194}
2195
2196#[cfg(test)]
2197mod acceptance_recording {
2198    //! The once-per-deployment recording of an operator's escape-hatch
2199    //! acceptance (issue #143, #437). In-file because the guards are
2200    //! private `AtomicBool`s on `Harness`, and per-`Harness` rather than a
2201    //! process static so each test's harness carries its own record.
2202
2203    use super::*;
2204    use crate::Migrations;
2205    use crate::config::MapConfig;
2206
2207    /// A surface-less public writer — the conservative-fallback shape,
2208    /// which the captcha leg satisfies through the runtime below.
2209    struct PubWriter;
2210
2211    impl Module for PubWriter {
2212        fn name(&self) -> &'static str {
2213            "writer"
2214        }
2215        fn version(&self) -> &'static str {
2216            "0.0.0-test"
2217        }
2218        fn requires(&self) -> &'static [Port] {
2219            &[]
2220        }
2221        fn public_writes(&self) -> bool {
2222            true
2223        }
2224        fn migrations(&self) -> Migrations {
2225            Migrations::default()
2226        }
2227        fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2228            Ok(())
2229        }
2230        fn router(&self, _: ModuleContext) -> Router {
2231            Router::new()
2232        }
2233    }
2234
2235    /// Advertises every port, so the captcha leg is satisfied and each
2236    /// test below is about the limiter leg alone.
2237    struct AllPorts;
2238
2239    impl Runtime for AllPorts {
2240        fn provides(&self) -> Vec<Port> {
2241            Port::ALL.to_vec()
2242        }
2243    }
2244
2245    fn harness() -> Harness {
2246        Harness::builder()
2247            .venture(
2248                Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2249            )
2250            .module(PubWriter)
2251            .runtime(AllPorts)
2252            .build()
2253            .expect("a development venture builds")
2254    }
2255
2256    fn deployed(unlimited: Option<&str>) -> MapConfig {
2257        let pairs = [("ENV", "production")].into_iter().chain(
2258            unlimited.map(|reason| (crate::route_policy::ALLOW_UNLIMITED_PUBLIC_ROUTES, reason)),
2259        );
2260        MapConfig::from_pairs(pairs)
2261    }
2262
2263    #[test]
2264    fn the_waiver_is_recorded_once_and_only_when_it_waived() {
2265        let harness = harness();
2266        let config = deployed(Some("issue #437: limiter binding pending"));
2267        assert!(
2268            harness
2269                .production_readiness_now(VentureEnv::Production, &config, false, true)
2270                .is_empty(),
2271            "the recorded acceptance serves"
2272        );
2273        assert!(
2274            harness
2275                .unlimited_acceptance_recorded
2276                .load(Ordering::Relaxed)
2277        );
2278
2279        // A later router build — on Workers there is one per request —
2280        // neither refuses nor re-records.
2281        assert!(
2282            harness
2283                .production_readiness_now(VentureEnv::Production, &config, false, true)
2284                .is_empty()
2285        );
2286    }
2287
2288    #[test]
2289    fn nothing_is_recorded_when_the_port_was_fine_all_along() {
2290        let harness = harness();
2291        let config = deployed(Some("unused: the limiter resolved"));
2292        assert!(
2293            harness
2294                .production_readiness_now(VentureEnv::Production, &config, true, true)
2295                .is_empty()
2296        );
2297        assert!(
2298            !harness
2299                .unlimited_acceptance_recorded
2300                .load(Ordering::Relaxed),
2301            "a waiver that removed no refusal is not a recorded acceptance"
2302        );
2303    }
2304
2305    #[test]
2306    fn a_refusing_deployment_records_no_acceptance_at_all() {
2307        let harness = harness();
2308        // The captcha waiver does not reach the limiter leg, so the gate
2309        // refuses — and an acceptance that left the deployment refusing
2310        // recorded nothing, for either key.
2311        let config = MapConfig::from_pairs([
2312            ("ENV", "production"),
2313            (
2314                crate::route_policy::ALLOW_UNPROTECTED_WRITES,
2315                "captcha pending",
2316            ),
2317        ]);
2318        assert_eq!(
2319            harness
2320                .production_readiness_now(VentureEnv::Production, &config, false, true)
2321                .len(),
2322            1
2323        );
2324        assert!(
2325            !harness
2326                .unprotected_acceptance_recorded
2327                .load(Ordering::Relaxed)
2328        );
2329        assert!(
2330            !harness
2331                .unlimited_acceptance_recorded
2332                .load(Ordering::Relaxed)
2333        );
2334    }
2335
2336    /// A webhook receiver that verifies its deliveries with the core HMAC
2337    /// scheme (issue #533): a signature writer that asks nothing of the
2338    /// Payments port — its secret key is the whole production gate.
2339    struct HmacWriter;
2340
2341    impl Module for HmacWriter {
2342        fn name(&self) -> &'static str {
2343            "webhooks"
2344        }
2345        fn version(&self) -> &'static str {
2346            "0.0.0-test"
2347        }
2348        fn requires(&self) -> &'static [Port] {
2349            &[]
2350        }
2351        fn public_writes(&self) -> bool {
2352            true
2353        }
2354        fn public_write_policy(&self) -> crate::route_policy::RoutePolicy {
2355            crate::route_policy::RoutePolicy::Signature
2356        }
2357        fn signature_verification(&self) -> crate::route_policy::SignatureVerification {
2358            crate::route_policy::SignatureVerification::Hmac {
2359                secret: "WEBHOOK_SECRET",
2360            }
2361        }
2362        fn migrations(&self) -> Migrations {
2363            Migrations::default()
2364        }
2365        fn validate_config(&self, _: &dyn Config) -> Result<(), ConfigError> {
2366            Ok(())
2367        }
2368        fn router(&self, _: ModuleContext) -> Router {
2369            Router::new()
2370        }
2371    }
2372
2373    #[test]
2374    fn a_non_payments_signature_venture_boots_on_its_webhook_secret_alone() {
2375        let harness = Harness::builder()
2376            .venture(
2377                Venture::new("test-venture", "test.example").cors_origins(["https://test.example"]),
2378            )
2379            .module(HmacWriter)
2380            .runtime(AllPorts)
2381            .build()
2382            .expect("a development venture builds");
2383        let deployed = MapConfig::from_pairs([("ENV", "production")]);
2384        let problems =
2385            harness.production_readiness_now(VentureEnv::Production, &deployed, true, true);
2386        assert_eq!(problems.len(), 1, "{problems:?}");
2387        assert!(
2388            problems[0].contains("WEBHOOKS_WEBHOOK_SECRET"),
2389            "{}",
2390            problems[0]
2391        );
2392        assert!(
2393            !problems[0].contains("Payments"),
2394            "a module verifying through the HMAC scheme is not the Payments leg's business: {}",
2395            problems[0]
2396        );
2397
2398        // The secret configured: the same venture is ready to serve.
2399        let ready = MapConfig::from_pairs([
2400            ("ENV", "production"),
2401            ("WEBHOOKS_WEBHOOK_SECRET", "whsec-test-dummy"),
2402        ]);
2403        assert!(
2404            harness
2405                .production_readiness_now(VentureEnv::Production, &ready, true, true)
2406                .is_empty()
2407        );
2408    }
2409}